Skip to content

F
flink

doujutun3207 / flink
与 Fork 源项目一致

从无法访问的项目Fork

通知 24
Star 0
Fork 0
F
flink

体验新版 GitCode，发现更多精彩内容 >>

提交 f3dee23b 编写于 作者: R r-pogalz 提交者： Fabian Hueske
浏览文件
操作

[FLINK-2106] [runtime] Add Outer Join drivers and Outer Merge strategies to Runtime

上级 24f7fa9e
隐藏空白更改
内联 并排
Showing with 1881 addition and 80 deletion
flink-runtime/src/main/java/org/apache/flink/runtime/operators/AbstractOuterJoinDriver.java 0 → 100644
浏览文件 @ f3dee23b
/*
* 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;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.FlatJoinFunction;
import org.apache.flink.api.common.typeutils.TypeComparator;
import org.apache.flink.api.common.typeutils.TypePairComparatorFactory;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
import org.apache.flink.runtime.memorymanager.MemoryManager;
import org.apache.flink.runtime.operators.util.JoinTaskIterator;
import org.apache.flink.runtime.operators.util.TaskConfig;
import org.apache.flink.util.Collector;
import org.apache.flink.util.MutableObjectIterator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* The abstract outer join driver implements the logic of an outer join operator at runtime. It instantiates a sort-merge based strategy to find
* joining pairs of records or joining records with null depending on the outer join type.
*
* @see FlatJoinFunction
*/
public abstract class AbstractOuterJoinDriver<IT1, IT2, OT> implements PactDriver<FlatJoinFunction<IT1, IT2, OT>, OT> {
protected static final Logger LOG = LoggerFactory.getLogger(AbstractOuterJoinDriver.class);
protected PactTaskContext<FlatJoinFunction<IT1, IT2, OT>, OT> taskContext;
protected volatile JoinTaskIterator<IT1, IT2, OT> outerJoinIterator; // the iterator that does the actual outer join
protected volatile boolean running;
// ------------------------------------------------------------------------
@Override
public void setup(PactTaskContext<FlatJoinFunction<IT1, IT2, OT>, OT> context) {
this.taskContext = context;
this.running = true;
}
@Override
public int getNumberOfInputs() {
return 2;
}
@Override
public Class<FlatJoinFunction<IT1, IT2, OT>> getStubType() {
@SuppressWarnings("unchecked")
final Class<FlatJoinFunction<IT1, IT2, OT>> clazz = (Class<FlatJoinFunction<IT1, IT2, OT>>) (Class<?>) FlatJoinFunction.class;
return clazz;
}
@Override
public int getNumberOfDriverComparators() {
return 2;
}
@Override
public void prepare() throws Exception {
final TaskConfig config = this.taskContext.getTaskConfig();
// obtain task manager's memory manager and I/O manager
final MemoryManager memoryManager = this.taskContext.getMemoryManager();
final IOManager ioManager = this.taskContext.getIOManager();
// set up memory and I/O parameters
final double fractionAvailableMemory = config.getRelativeMemoryDriver();
final int numPages = memoryManager.computeNumberOfPages(fractionAvailableMemory);
final DriverStrategy ls = config.getDriverStrategy();
final MutableObjectIterator<IT1> in1 = this.taskContext.getInput(0);
final MutableObjectIterator<IT2> in2 = this.taskContext.getInput(1);
// get serializers and comparators
final TypeSerializer<IT1> serializer1 = this.taskContext.<IT1>getInputSerializer(0).getSerializer();
final TypeSerializer<IT2> serializer2 = this.taskContext.<IT2>getInputSerializer(1).getSerializer();
final TypeComparator<IT1> comparator1 = this.taskContext.getDriverComparator(0);
final TypeComparator<IT2> comparator2 = this.taskContext.getDriverComparator(1);
final TypePairComparatorFactory<IT1, IT2> pairComparatorFactory = config.getPairComparatorFactory(this.taskContext.getUserCodeClassLoader());
if (pairComparatorFactory == null) {
throw new Exception("Missing pair comparator factory for outer join driver");
}
ExecutionConfig executionConfig = taskContext.getExecutionConfig();
boolean objectReuseEnabled = executionConfig.isObjectReuseEnabled();
if (LOG.isDebugEnabled()) {
LOG.debug("Outer Join Driver object reuse: " + (objectReuseEnabled ? "ENABLED" : "DISABLED") + ".");
}
// create and return outer join iterator according to provided local strategy.
if (objectReuseEnabled) {
this.outerJoinIterator = getReusingOuterJoinIterator(
ls,
in1,
in2,
serializer1,
comparator1,
serializer2,
comparator2,
pairComparatorFactory,
memoryManager,
ioManager,
numPages
);
} else {
this.outerJoinIterator = getNonReusingOuterJoinIterator(
ls,
in1,
in2,
serializer1,
comparator1,
serializer2,
comparator2,
pairComparatorFactory,
memoryManager,
ioManager,
numPages
);
}
this.outerJoinIterator.open();
if (LOG.isDebugEnabled()) {
LOG.debug(this.taskContext.formatLogString("outer join task iterator ready."));
}
}
@Override
public void run() throws Exception {
final FlatJoinFunction<IT1, IT2, OT> joinStub = this.taskContext.getStub();
final Collector<OT> collector = this.taskContext.getOutputCollector();
final JoinTaskIterator<IT1, IT2, OT> outerJoinIterator = this.outerJoinIterator;
while (this.running && outerJoinIterator.callWithNextKey(joinStub, collector)) ;
}
@Override
public void cleanup() throws Exception {
if (this.outerJoinIterator != null) {
this.outerJoinIterator.close();
this.outerJoinIterator = null;
}
}
@Override
public void cancel() {
this.running = false;
if (this.outerJoinIterator != null) {
this.outerJoinIterator.abort();
}
}
protected abstract JoinTaskIterator<IT1, IT2, OT> getReusingOuterJoinIterator(
DriverStrategy driverStrategy,
MutableObjectIterator<IT1> in1,
MutableObjectIterator<IT2> in2,
TypeSerializer<IT1> serializer1,
TypeComparator<IT1> comparator1,
TypeSerializer<IT2> serializer2,
TypeComparator<IT2> comparator2,
TypePairComparatorFactory<IT1, IT2> pairComparatorFactory,
MemoryManager memoryManager,
IOManager ioManager,
int numPages
) throws Exception;
protected abstract JoinTaskIterator<IT1, IT2, OT> getNonReusingOuterJoinIterator(
DriverStrategy driverStrategy,
MutableObjectIterator<IT1> in1,
MutableObjectIterator<IT2> in2,
TypeSerializer<IT1> serializer1,
TypeComparator<IT1> comparator1,
TypeSerializer<IT2> serializer2,
TypeComparator<IT2> comparator2,
TypePairComparatorFactory<IT1, IT2> pairComparatorFactory,
MemoryManager memoryManager,
IOManager ioManager,
int numPages
) throws Exception;
}
flink-runtime/src/main/java/org/apache/flink/runtime/operators/DriverStrategy.java
浏览文件 @ f3dee23b
......@@ -74,6 +74,12 @@ public enum DriverStrategy {
// both inputs are merged, but materialized to the side for block-nested-loop-join among values with equal key
MERGE(JoinDriver.class, null, MATERIALIZING, MATERIALIZING, 2),
LEFT_OUTER_MERGE(LeftOuterJoinDriver.class, null, MATERIALIZING, MATERIALIZING, 2),
RIGHT_OUTER_MERGE(RightOuterJoinDriver.class, null, MATERIALIZING, MATERIALIZING, 2),
FULL_OUTER_MERGE(FullOuterJoinDriver.class, null, MATERIALIZING, MATERIALIZING, 2),
// co-grouping inputs
CO_GROUP(CoGroupDriver.class, null, PIPELINED, PIPELINED, 2),
// python-cogroup
......
flink-runtime/src/main/java/org/apache/flink/runtime/operators/FullOuterJoinDriver.java 0 → 100644
浏览文件 @ f3dee23b
/*
* 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;
import org.apache.flink.api.common.typeutils.TypeComparator;
import org.apache.flink.api.common.typeutils.TypePairComparatorFactory;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
import org.apache.flink.runtime.memorymanager.MemoryManager;
import org.apache.flink.runtime.operators.sort.AbstractMergeOuterJoinIterator.OuterJoinType;
import org.apache.flink.runtime.operators.sort.NonReusingMergeOuterJoinIterator;
import org.apache.flink.runtime.operators.sort.ReusingMergeOuterJoinIterator;
import org.apache.flink.runtime.operators.util.JoinTaskIterator;
import org.apache.flink.util.MutableObjectIterator;
/**
* The full outer join driver implements the logic of an outer join operator at runtime. It instantiates a sort-merge based strategy to find
* joining pairs of records or joins records with null if no match is found.
*/
public class FullOuterJoinDriver<IT1, IT2, OT> extends AbstractOuterJoinDriver<IT1, IT2, OT> {
@Override
protected JoinTaskIterator<IT1, IT2, OT> getReusingOuterJoinIterator(
DriverStrategy driverStrategy,
MutableObjectIterator<IT1> in1,
MutableObjectIterator<IT2> in2,
TypeSerializer<IT1> serializer1,
TypeComparator<IT1> comparator1,
TypeSerializer<IT2> serializer2,
TypeComparator<IT2> comparator2,
TypePairComparatorFactory<IT1, IT2> pairComparatorFactory,
MemoryManager memoryManager,
IOManager ioManager,
int numPages
) throws Exception {
switch (driverStrategy) {
case FULL_OUTER_MERGE:
return new ReusingMergeOuterJoinIterator<>(
OuterJoinType.FULL,
in1,
in2,
serializer1,
comparator1,
serializer2,
comparator2,
pairComparatorFactory.createComparator12(comparator1, comparator2),
memoryManager,
ioManager,
numPages,
super.taskContext.getOwningNepheleTask()
);
default:
throw new Exception("Unsupported driver strategy for full outer join driver: " + driverStrategy.name());
}
}
@Override
protected JoinTaskIterator<IT1, IT2, OT> getNonReusingOuterJoinIterator(
DriverStrategy driverStrategy,
MutableObjectIterator<IT1> in1,
MutableObjectIterator<IT2> in2,
TypeSerializer<IT1> serializer1,
TypeComparator<IT1> comparator1,
TypeSerializer<IT2> serializer2,
TypeComparator<IT2> comparator2,
TypePairComparatorFactory<IT1, IT2> pairComparatorFactory,
MemoryManager memoryManager,
IOManager ioManager,
int numPages
) throws Exception {
switch (driverStrategy) {
case FULL_OUTER_MERGE:
return new NonReusingMergeOuterJoinIterator<>(
OuterJoinType.FULL,
in1,
in2,
serializer1,
comparator1,
serializer2,
comparator2,
pairComparatorFactory.createComparator12(comparator1, comparator2),
memoryManager,
ioManager,
numPages,
super.taskContext.getOwningNepheleTask()
);
default:
throw new Exception("Unsupported driver strategy for full outer join driver: " + driverStrategy.name());
}
}
}
\ No newline at end of file
flink-runtime/src/main/java/org/apache/flink/runtime/operators/LeftOuterJoinDriver.java 0 → 100644
浏览文件 @ f3dee23b
/*
* 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;
import org.apache.flink.api.common.typeutils.TypeComparator;
import org.apache.flink.api.common.typeutils.TypePairComparatorFactory;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
import org.apache.flink.runtime.memorymanager.MemoryManager;
import org.apache.flink.runtime.operators.sort.AbstractMergeOuterJoinIterator.OuterJoinType;
import org.apache.flink.runtime.operators.sort.NonReusingMergeOuterJoinIterator;
import org.apache.flink.runtime.operators.sort.ReusingMergeOuterJoinIterator;
import org.apache.flink.runtime.operators.util.JoinTaskIterator;
import org.apache.flink.util.MutableObjectIterator;
/**
* The left outer join driver implements the logic of an outer join operator at runtime. It instantiates a sort-merge based strategy to find
* joining pairs of records or joins records from the left side with null if no match is found.
*/
public class LeftOuterJoinDriver<IT1, IT2, OT> extends AbstractOuterJoinDriver<IT1, IT2, OT> {
@Override
protected JoinTaskIterator<IT1, IT2, OT> getReusingOuterJoinIterator(
DriverStrategy driverStrategy,
MutableObjectIterator<IT1> in1,
MutableObjectIterator<IT2> in2,
TypeSerializer<IT1> serializer1,
TypeComparator<IT1> comparator1,
TypeSerializer<IT2> serializer2,
TypeComparator<IT2> comparator2,
TypePairComparatorFactory<IT1, IT2> pairComparatorFactory,
MemoryManager memoryManager,
IOManager ioManager,
int numPages
) throws Exception {
switch (driverStrategy) {
case LEFT_OUTER_MERGE:
return new ReusingMergeOuterJoinIterator<>(
OuterJoinType.LEFT,
in1,
in2,
serializer1,
comparator1,
serializer2,
comparator2,
pairComparatorFactory.createComparator12(comparator1, comparator2),
memoryManager,
ioManager,
numPages,
super.taskContext.getOwningNepheleTask()
);
default:
throw new Exception("Unsupported driver strategy for left outer join driver: " + driverStrategy.name());
}
}
@Override
protected JoinTaskIterator<IT1, IT2, OT> getNonReusingOuterJoinIterator(
DriverStrategy driverStrategy,
MutableObjectIterator<IT1> in1,
MutableObjectIterator<IT2> in2,
TypeSerializer<IT1> serializer1,
TypeComparator<IT1> comparator1,
TypeSerializer<IT2> serializer2,
TypeComparator<IT2> comparator2,
TypePairComparatorFactory<IT1, IT2> pairComparatorFactory,
MemoryManager memoryManager,
IOManager ioManager,
int numPages
) throws Exception {
switch (driverStrategy) {
case LEFT_OUTER_MERGE:
return new NonReusingMergeOuterJoinIterator<>(
OuterJoinType.LEFT,
in1,
in2,
serializer1,
comparator1,
serializer2,
comparator2,
pairComparatorFactory.createComparator12(comparator1, comparator2),
memoryManager,
ioManager,
numPages,
super.taskContext.getOwningNepheleTask()
);
default:
throw new Exception("Unsupported driver strategy for left outer join driver: " + driverStrategy.name());
}
}
}
\ No newline at end of file
flink-runtime/src/main/java/org/apache/flink/runtime/operators/RightOuterJoinDriver.java 0 → 100644
浏览文件 @ f3dee23b
/*
* 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;
import org.apache.flink.api.common.typeutils.TypeComparator;
import org.apache.flink.api.common.typeutils.TypePairComparatorFactory;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
import org.apache.flink.runtime.memorymanager.MemoryManager;
import org.apache.flink.runtime.operators.sort.AbstractMergeOuterJoinIterator.OuterJoinType;
import org.apache.flink.runtime.operators.sort.NonReusingMergeOuterJoinIterator;
import org.apache.flink.runtime.operators.sort.ReusingMergeOuterJoinIterator;
import org.apache.flink.runtime.operators.util.JoinTaskIterator;
import org.apache.flink.util.MutableObjectIterator;
/**
* The right outer join driver implements the logic of an outer join operator at runtime. It instantiates a sort-merge based strategy to find
* joining pairs of records or joins records from the right side with null if no match is found.
*/
public class RightOuterJoinDriver<IT1, IT2, OT> extends AbstractOuterJoinDriver<IT1, IT2, OT> {
@Override
protected JoinTaskIterator<IT1, IT2, OT> getReusingOuterJoinIterator(
DriverStrategy driverStrategy,
MutableObjectIterator<IT1> in1,
MutableObjectIterator<IT2> in2,
TypeSerializer<IT1> serializer1,
TypeComparator<IT1> comparator1,
TypeSerializer<IT2> serializer2,
TypeComparator<IT2> comparator2,
TypePairComparatorFactory<IT1, IT2> pairComparatorFactory,
MemoryManager memoryManager,
IOManager ioManager,
int numPages
) throws Exception {
switch (driverStrategy) {
case RIGHT_OUTER_MERGE:
return new ReusingMergeOuterJoinIterator<>(
OuterJoinType.RIGHT,
in1,
in2,
serializer1,
comparator1,
serializer2,
comparator2,
pairComparatorFactory.createComparator12(comparator1, comparator2),
memoryManager,
ioManager,
numPages,
super.taskContext.getOwningNepheleTask()
);
default:
throw new Exception("Unsupported driver strategy for right outer join driver: " + driverStrategy.name());
}
}
@Override
protected JoinTaskIterator<IT1, IT2, OT> getNonReusingOuterJoinIterator(
DriverStrategy driverStrategy,
MutableObjectIterator<IT1> in1,
MutableObjectIterator<IT2> in2,
TypeSerializer<IT1> serializer1,
TypeComparator<IT1> comparator1,
TypeSerializer<IT2> serializer2,
TypeComparator<IT2> comparator2,
TypePairComparatorFactory<IT1, IT2> pairComparatorFactory,
MemoryManager memoryManager,
IOManager ioManager,
int numPages
) throws Exception {
switch (driverStrategy) {
case RIGHT_OUTER_MERGE:
return new NonReusingMergeOuterJoinIterator<>(
OuterJoinType.RIGHT,
in1,
in2,
serializer1,
comparator1,
serializer2,
comparator2,
pairComparatorFactory.createComparator12(comparator1, comparator2),
memoryManager,
ioManager,
numPages,
super.taskContext.getOwningNepheleTask()
);
default:
throw new Exception("Unsupported driver strategy for right outer join driver: " + driverStrategy.name());
}
}
}
\ No newline at end of file
flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/AbstractMergeIterator.java
浏览文件 @ f3dee23b
......@@ -87,7 +87,7 @@ public abstract class AbstractMergeIterator<T1, T2, O> implements JoinTaskIterat
this.iterator2 = createKeyGroupedIterator(input2, serializer2, comparator2.duplicate());
final int numPagesForSpiller = numMemoryPages > 20 ? 2 : 1;
this.blockIt = new NonReusingBlockResettableIterator<T2>(this.memoryManager, this.serializer2,
this.blockIt = new NonReusingBlockResettableIterator<>(this.memoryManager, this.serializer2,
(numMemoryPages - numPagesForSpiller), parentTask);
this.memoryForSpillingIterator = memoryManager.allocatePages(parentTask, numPagesForSpiller);
}
......@@ -267,8 +267,9 @@ public abstract class AbstractMergeIterator<T1, T2, O> implements JoinTaskIterat
if (spillingRequired) {
// more data than would fit into one block. we need to wrap the other side in a spilling iterator
// create spilling iterator on first input
spillIt = new SpillingResettableIterator<T1>(spillVals, this.serializer1,
this.memoryManager, this.ioManager, this.memoryForSpillingIterator);
spillIt = new SpillingResettableIterator<>(
spillVals, this.serializer1, this.memoryManager, this.ioManager, this.memoryForSpillingIterator
);
leftSideIter = spillIt;
spillIt.open();
......
flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/AbstractMergeOuterJoinIterator.java
浏览文件 @ f3dee23b
......@@ -37,7 +37,7 @@ import java.util.Iterator;
*/
public abstract class AbstractMergeOuterJoinIterator<T1, T2, O> extends AbstractMergeIterator<T1, T2, O> {
public static enum OuterJoinType {LEFT, RIGHT, FULL}
public enum OuterJoinType {LEFT, RIGHT, FULL}
private final OuterJoinType outerJoinType;
......
flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/NonReusingMergeOuterJoinIterator.java
浏览文件 @ f3dee23b
......@@ -48,7 +48,7 @@ public class NonReusingMergeOuterJoinIterator<T1, T2, O> extends AbstractMergeOu
@Override
protected <T> KeyGroupedIterator<T> createKeyGroupedIterator(MutableObjectIterator<T> input, TypeSerializer<T> serializer, TypeComparator<T> comparator) {
return new NonReusingKeyGroupedIterator<T>(input, comparator);
return new NonReusingKeyGroupedIterator<>(input, comparator);
}
@Override
......
flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/ReusingMergeOuterJoinIterator.java
浏览文件 @ f3dee23b
......@@ -53,7 +53,7 @@ public class ReusingMergeOuterJoinIterator<T1, T2, O> extends AbstractMergeOuter
@Override
protected <T> KeyGroupedIterator<T> createKeyGroupedIterator(MutableObjectIterator<T> input, TypeSerializer<T> serializer, TypeComparator<T> comparator) {
return new ReusingKeyGroupedIterator<T>(input, serializer, comparator);
return new ReusingKeyGroupedIterator<>(input, serializer, comparator);
}
@Override
......
flink-runtime/src/test/java/org/apache/flink/runtime/operators/AbstractOuterJoinTaskExternalITCase.java 0 → 100644
浏览文件 @ f3dee23b
/*
* 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;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.FlatJoinFunction;
import org.apache.flink.api.common.typeutils.TypeComparator;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.base.IntComparator;
import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.runtime.RuntimePairComparatorFactory;
import org.apache.flink.api.java.typeutils.runtime.TupleComparator;
import org.apache.flink.api.java.typeutils.runtime.TupleSerializer;
import org.apache.flink.runtime.operators.testutils.BinaryOperatorTestBase;
import org.apache.flink.runtime.operators.testutils.UniformIntTupleGenerator;
import org.apache.flink.util.Collector;
import org.junit.Assert;
import org.junit.Test;
public abstract class AbstractOuterJoinTaskExternalITCase extends BinaryOperatorTestBase<FlatJoinFunction<Tuple2<Integer, Integer>,
Tuple2<Integer, Integer>, Tuple2<Integer, Integer>>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> {
private static final long HASH_MEM = 4 * 1024 * 1024;
private static final long SORT_MEM = 3 * 1024 * 1024;
private static final long BNLJN_MEM = 10 * PAGE_SIZE;
private final double bnljn_frac;
@SuppressWarnings("unchecked")
private final TypeComparator<Tuple2<Integer, Integer>> comparator1 = new TupleComparator<>(
new int[]{0},
new TypeComparator<?>[]{new IntComparator(true)},
new TypeSerializer<?>[]{IntSerializer.INSTANCE}
);
@SuppressWarnings("unchecked")
private final TypeComparator<Tuple2<Integer, Integer>> comparator2 = new TupleComparator<>(
new int[]{0},
new TypeComparator<?>[]{new IntComparator(true)},
new TypeSerializer<?>[]{IntSerializer.INSTANCE}
);
@SuppressWarnings("unchecked")
private final TypeSerializer<Tuple2<Integer, Integer>> serializer = new TupleSerializer<>(
(Class<Tuple2<Integer, Integer>>) (Class<?>) Tuple2.class,
new TypeSerializer<?>[]{IntSerializer.INSTANCE, IntSerializer.INSTANCE}
);
private final CountingOutputCollector<Tuple2<Integer, Integer>> output = new CountingOutputCollector<>();
private final DriverStrategy driverStrategy;
public AbstractOuterJoinTaskExternalITCase(ExecutionConfig config, DriverStrategy driverStrategy) {
super(config, HASH_MEM, 2, SORT_MEM);
bnljn_frac = (double) BNLJN_MEM / this.getMemoryManager().getMemorySize();
this.driverStrategy = driverStrategy;
}
@Test
public void testExternalSortOuterJoinTask() throws Exception {
final int keyCnt1 = 16384 * 4;
final int valCnt1 = 2;
final int keyCnt2 = 8192;
final int valCnt2 = 4 * 2;
final int expCnt = calculateExpectedCount(keyCnt1, valCnt1, keyCnt2, valCnt2);
setOutput(this.output);
addDriverComparator(this.comparator1);
addDriverComparator(this.comparator2);
getTaskConfig().setDriverPairComparator(new RuntimePairComparatorFactory());
getTaskConfig().setDriverStrategy(driverStrategy);
getTaskConfig().setRelativeMemoryDriver(bnljn_frac);
setNumFileHandlesForSort(4);
final AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> testTask = getOuterJoinDriver();
addInputSorted(new UniformIntTupleGenerator(keyCnt1, valCnt1, false), serializer, this.comparator1.duplicate());
addInputSorted(new UniformIntTupleGenerator(keyCnt2, valCnt2, false), serializer, this.comparator2.duplicate());
testDriver(testTask, MockJoinStub.class);
Assert.assertEquals("Wrong result set size.", expCnt, this.output.getNumberOfRecords());
}
protected abstract int calculateExpectedCount(int keyCnt1, int valCnt1, int keyCnt2, int valCnt2);
protected abstract AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> getOuterJoinDriver();
// =================================================================================================
@SuppressWarnings("serial")
public static final class MockJoinStub implements FlatJoinFunction<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> {
@Override
public void join(Tuple2<Integer, Integer> first, Tuple2<Integer, Integer> second, Collector<Tuple2<Integer, Integer>> out) throws Exception {
out.collect(first != null ? first : second);
}
}
}
flink-runtime/src/test/java/org/apache/flink/runtime/operators/AbstractOuterJoinTaskTest.java 0 → 100644
浏览文件 @ f3dee23b
/*
* 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;
import com.google.common.base.Throwables;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.FlatJoinFunction;
import org.apache.flink.api.common.typeutils.TypeComparator;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.base.IntComparator;
import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.runtime.RuntimePairComparatorFactory;
import org.apache.flink.api.java.typeutils.runtime.TupleComparator;
import org.apache.flink.api.java.typeutils.runtime.TupleSerializer;
import org.apache.flink.runtime.operators.testutils.BinaryOperatorTestBase;
import org.apache.flink.runtime.operators.testutils.DelayingIterator;
import org.apache.flink.runtime.operators.testutils.DiscardingOutputCollector;
import org.apache.flink.runtime.operators.testutils.ExpectedTestException;
import org.apache.flink.runtime.operators.testutils.InfiniteIntTupleIterator;
import org.apache.flink.runtime.operators.testutils.UniformIntTupleGenerator;
import org.apache.flink.util.Collector;
import org.junit.Assert;
import org.junit.Test;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicReference;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.fail;
public abstract class AbstractOuterJoinTaskTest extends BinaryOperatorTestBase<FlatJoinFunction<Tuple2<Integer, Integer>,
Tuple2<Integer, Integer>, Tuple2<Integer, Integer>>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> {
private static final long HASH_MEM = 6 * 1024 * 1024;
private static final long SORT_MEM = 3 * 1024 * 1024;
private static final int NUM_SORTER = 2;
private static final long BNLJN_MEM = 10 * PAGE_SIZE;
private final double bnljn_frac;
private final DriverStrategy driverStrategy;
@SuppressWarnings("unchecked")
private final TypeComparator<Tuple2<Integer, Integer>> comparator1 = new TupleComparator<>(
new int[]{0},
new TypeComparator<?>[]{new IntComparator(true)},
new TypeSerializer<?>[]{IntSerializer.INSTANCE}
);
@SuppressWarnings("unchecked")
private final TypeComparator<Tuple2<Integer, Integer>> comparator2 = new TupleComparator<>(
new int[]{0},
new TypeComparator<?>[]{new IntComparator(true)},
new TypeSerializer<?>[]{IntSerializer.INSTANCE}
);
private final List<Tuple2<Integer, Integer>> outList = new ArrayList<>();
@SuppressWarnings("unchecked")
private final TypeSerializer<Tuple2<Integer, Integer>> serializer = new TupleSerializer<>(
(Class<Tuple2<Integer, Integer>>) (Class<?>) Tuple2.class,
new TypeSerializer<?>[]{IntSerializer.INSTANCE, IntSerializer.INSTANCE}
);
public AbstractOuterJoinTaskTest(ExecutionConfig config, DriverStrategy driverStrategy) {
super(config, HASH_MEM, NUM_SORTER, SORT_MEM);
bnljn_frac = (double) BNLJN_MEM / this.getMemoryManager().getMemorySize();
this.driverStrategy = driverStrategy;
}
@Test
public void testSortBoth1OuterJoinTask() throws Exception {
final int keyCnt1 = 20;
final int valCnt1 = 1;
final int keyCnt2 = 10;
final int valCnt2 = 2;
testSortBothOuterJoinTask(keyCnt1, valCnt1, keyCnt2, valCnt2);
}
@Test
public void testSortBoth2OuterJoinTask() throws Exception {
final int keyCnt1 = 20;
final int valCnt1 = 1;
final int keyCnt2 = 20;
final int valCnt2 = 1;
testSortBothOuterJoinTask(keyCnt1, valCnt1, keyCnt2, valCnt2);
}
@Test
public void testSortBoth3OuterJoinTask() throws Exception {
int keyCnt1 = 20;
int valCnt1 = 1;
int keyCnt2 = 20;
int valCnt2 = 20;
testSortBothOuterJoinTask(keyCnt1, valCnt1, keyCnt2, valCnt2);
}
@Test
public void testSortBoth4OuterJoinTask() throws Exception {
int keyCnt1 = 20;
int valCnt1 = 20;
int keyCnt2 = 20;
int valCnt2 = 1;
testSortBothOuterJoinTask(keyCnt1, valCnt1, keyCnt2, valCnt2);
}
@Test
public void testSortBoth5OuterJoinTask() throws Exception {
int keyCnt1 = 20;
int valCnt1 = 20;
int keyCnt2 = 20;
int valCnt2 = 20;
testSortBothOuterJoinTask(keyCnt1, valCnt1, keyCnt2, valCnt2);
}
@Test
public void testSortBoth6OuterJoinTask() throws Exception {
int keyCnt1 = 10;
int valCnt1 = 1;
int keyCnt2 = 20;
int valCnt2 = 2;
testSortBothOuterJoinTask(keyCnt1, valCnt1, keyCnt2, valCnt2);
}
private void testSortBothOuterJoinTask(int keyCnt1, int valCnt1, int keyCnt2, int valCnt2) throws Exception {
setOutput(this.outList, this.serializer);
addDriverComparator(this.comparator1);
addDriverComparator(this.comparator2);
getTaskConfig().setDriverPairComparator(new RuntimePairComparatorFactory());
getTaskConfig().setDriverStrategy(this.driverStrategy);
getTaskConfig().setRelativeMemoryDriver(this.bnljn_frac);
setNumFileHandlesForSort(4);
final AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> testTask = getOuterJoinDriver();
addInputSorted(new UniformIntTupleGenerator(keyCnt1, valCnt1, false), this.serializer, this.comparator1.duplicate());
addInputSorted(new UniformIntTupleGenerator(keyCnt2, valCnt2, false), this.serializer, this.comparator2.duplicate());
testDriver(testTask, MockJoinStub.class);
final int expCnt = calculateExpectedCount(keyCnt1, valCnt1, keyCnt2, valCnt2);
Assert.assertTrue("Result set size was " + this.outList.size() + ". Expected was " + expCnt, this.outList.size() == expCnt);
this.outList.clear();
}
@Test
public void testSortFirstOuterJoinTask() throws Exception {
int keyCnt1 = 20;
int valCnt1 = 20;
int keyCnt2 = 20;
int valCnt2 = 20;
setOutput(this.outList, this.serializer);
addDriverComparator(this.comparator1);
addDriverComparator(this.comparator2);
getTaskConfig().setDriverPairComparator(new RuntimePairComparatorFactory());
getTaskConfig().setDriverStrategy(this.driverStrategy);
getTaskConfig().setRelativeMemoryDriver(this.bnljn_frac);
setNumFileHandlesForSort(4);
final AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> testTask = getOuterJoinDriver();
addInputSorted(new UniformIntTupleGenerator(keyCnt1, valCnt1, false), this.serializer, this.comparator1.duplicate());
addInput(new UniformIntTupleGenerator(keyCnt2, valCnt2, true), this.serializer);
testDriver(testTask, MockJoinStub.class);
final int expCnt = calculateExpectedCount(keyCnt1, valCnt1, keyCnt2, valCnt2);
Assert.assertTrue("Result set size was " + this.outList.size() + ". Expected was " + expCnt, this.outList.size() == expCnt);
this.outList.clear();
}
@Test
public void testSortSecondOuterJoinTask() throws Exception {
int keyCnt1 = 20;
int valCnt1 = 20;
int keyCnt2 = 20;
int valCnt2 = 20;
setOutput(this.outList, this.serializer);
addDriverComparator(this.comparator1);
addDriverComparator(this.comparator2);
getTaskConfig().setDriverPairComparator(new RuntimePairComparatorFactory());
getTaskConfig().setDriverStrategy(this.driverStrategy);
getTaskConfig().setRelativeMemoryDriver(this.bnljn_frac);
setNumFileHandlesForSort(4);
final AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> testTask = getOuterJoinDriver();
addInput(new UniformIntTupleGenerator(keyCnt1, valCnt1, true), this.serializer);
addInputSorted(new UniformIntTupleGenerator(keyCnt2, valCnt2, false), this.serializer, this.comparator2.duplicate());
testDriver(testTask, MockJoinStub.class);
final int expCnt = calculateExpectedCount(keyCnt1, valCnt1, keyCnt2, valCnt2);
Assert.assertTrue("Result set size was " + this.outList.size() + ". Expected was " + expCnt, this.outList.size() == expCnt);
this.outList.clear();
}
@Test
public void testMergeOuterJoinTask() throws Exception {
int keyCnt1 = 20;
int valCnt1 = 20;
int keyCnt2 = 20;
int valCnt2 = 20;
setOutput(this.outList, this.serializer);
addDriverComparator(this.comparator1);
addDriverComparator(this.comparator2);
getTaskConfig().setDriverPairComparator(new RuntimePairComparatorFactory());
getTaskConfig().setDriverStrategy(this.driverStrategy);
getTaskConfig().setRelativeMemoryDriver(this.bnljn_frac);
setNumFileHandlesForSort(4);
final AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> testTask = getOuterJoinDriver();
addInput(new UniformIntTupleGenerator(keyCnt1, valCnt1, true), this.serializer);
addInput(new UniformIntTupleGenerator(keyCnt2, valCnt2, true), this.serializer);
testDriver(testTask, MockJoinStub.class);
final int expCnt = calculateExpectedCount(keyCnt1, valCnt1, keyCnt2, valCnt2);
Assert.assertTrue("Result set size was " + this.outList.size() + ". Expected was " + expCnt, this.outList.size() == expCnt);
this.outList.clear();
}
@Test(expected = ExpectedTestException.class)
public void testFailingOuterJoinTask() throws Exception {
int keyCnt1 = 20;
int valCnt1 = 20;
int keyCnt2 = 20;
int valCnt2 = 20;
setOutput(new DiscardingOutputCollector<Tuple2<Integer, Integer>>());
addDriverComparator(this.comparator1);
addDriverComparator(this.comparator2);
getTaskConfig().setDriverPairComparator(new RuntimePairComparatorFactory());
getTaskConfig().setDriverStrategy(this.driverStrategy);
getTaskConfig().setRelativeMemoryDriver(this.bnljn_frac);
setNumFileHandlesForSort(4);
final AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> testTask = getOuterJoinDriver();
addInput(new UniformIntTupleGenerator(keyCnt1, valCnt1, true), this.serializer);
addInput(new UniformIntTupleGenerator(keyCnt2, valCnt2, true), this.serializer);
testDriver(testTask, MockFailingJoinStub.class);
}
@Test
public void testCancelOuterJoinTaskWhileSort1() throws Exception {
setOutput(new DiscardingOutputCollector<Tuple2<Integer, Integer>>());
addDriverComparator(this.comparator1);
addDriverComparator(this.comparator2);
getTaskConfig().setDriverPairComparator(new RuntimePairComparatorFactory());
getTaskConfig().setDriverStrategy(this.driverStrategy);
getTaskConfig().setRelativeMemoryDriver(this.bnljn_frac);
setNumFileHandlesForSort(4);
final AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> testTask = getOuterJoinDriver();
addInputSorted(new DelayingIterator<>(new InfiniteIntTupleIterator(), 100), this.serializer, this.comparator1.duplicate());
addInput(new DelayingIterator<>(new InfiniteIntTupleIterator(), 100), this.serializer);
final AtomicReference<Throwable> error = new AtomicReference<>();
final Thread taskRunner = new Thread("Task runner for testCancelOuterJoinTaskWhileSort1()") {
@Override
public void run() {
try {
testDriver(testTask, MockJoinStub.class);
} catch (Throwable t) {
error.set(t);
}
}
};
taskRunner.start();
Thread.sleep(1000);
cancel();
taskRunner.interrupt();
taskRunner.join(60000);
assertFalse("Task thread did not finish within 60 seconds", taskRunner.isAlive());
final Throwable taskError = error.get();
if (taskError != null) {
fail("Error in task while canceling:\n" + Throwables.getStackTraceAsString(taskError));
}
}
@Test
public void testCancelOuterJoinTaskWhileSort2() throws Exception {
setOutput(new DiscardingOutputCollector<Tuple2<Integer, Integer>>());
addDriverComparator(this.comparator1);
addDriverComparator(this.comparator2);
getTaskConfig().setDriverPairComparator(new RuntimePairComparatorFactory());
getTaskConfig().setDriverStrategy(this.driverStrategy);
getTaskConfig().setRelativeMemoryDriver(this.bnljn_frac);
setNumFileHandlesForSort(4);
final AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> testTask = getOuterJoinDriver();
addInput(new DelayingIterator<>(new InfiniteIntTupleIterator(), 1), this.serializer);
addInputSorted(new DelayingIterator<>(new InfiniteIntTupleIterator(), 1), this.serializer, this.comparator2.duplicate());
final AtomicReference<Throwable> error = new AtomicReference<>();
final Thread taskRunner = new Thread("Task runner for testCancelOuterJoinTaskWhileSort2()") {
@Override
public void run() {
try {
testDriver(testTask, MockJoinStub.class);
} catch (Throwable t) {
error.set(t);
}
}
};
taskRunner.start();
Thread.sleep(1000);
cancel();
taskRunner.interrupt();
taskRunner.join(60000);
assertFalse("Task thread did not finish within 60 seconds", taskRunner.isAlive());
final Throwable taskError = error.get();
if (taskError != null) {
fail("Error in task while canceling:\n" + Throwables.getStackTraceAsString(taskError));
}
}
@Test
public void testCancelOuterJoinTaskWhileRunning() throws Exception {
setOutput(new DiscardingOutputCollector<Tuple2<Integer, Integer>>());
addDriverComparator(this.comparator1);
addDriverComparator(this.comparator2);
getTaskConfig().setDriverPairComparator(new RuntimePairComparatorFactory());
getTaskConfig().setDriverStrategy(driverStrategy);
getTaskConfig().setRelativeMemoryDriver(bnljn_frac);
setNumFileHandlesForSort(4);
final AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> testTask = getOuterJoinDriver();
addInput(new DelayingIterator<>(new InfiniteIntTupleIterator(), 100), this.serializer);
addInput(new DelayingIterator<>(new InfiniteIntTupleIterator(), 100), this.serializer);
final AtomicReference<Throwable> error = new AtomicReference<>();
final Thread taskRunner = new Thread("Task runner for testCancelOuterJoinTaskWhileRunning()") {
@Override
public void run() {
try {
testDriver(testTask, MockJoinStub.class);
} catch (Throwable t) {
error.set(t);
}
}
};
taskRunner.start();
Thread.sleep(1000);
cancel();
taskRunner.interrupt();
taskRunner.join(60000);
assertFalse("Task thread did not finish within 60 seconds", taskRunner.isAlive());
final Throwable taskError = error.get();
if (taskError != null) {
fail("Error in task while canceling:\n" + Throwables.getStackTraceAsString(taskError));
}
}
protected abstract AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> getOuterJoinDriver();
protected abstract int calculateExpectedCount(int keyCnt1, int valCnt1, int keyCnt2, int valCnt2);
// =================================================================================================
@SuppressWarnings("serial")
public static final class MockJoinStub implements FlatJoinFunction<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> {
@Override
public void join(Tuple2<Integer, Integer> first, Tuple2<Integer, Integer> second, Collector<Tuple2<Integer, Integer>> out) throws Exception {
out.collect(first != null ? first : second);
}
}
@SuppressWarnings("serial")
public static final class MockFailingJoinStub implements FlatJoinFunction<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> {
private int cnt = 0;
@Override
public void join(Tuple2<Integer, Integer> first, Tuple2<Integer, Integer> second, Collector<Tuple2<Integer, Integer>> out) throws Exception {
if (++this.cnt >= 10) {
throw new ExpectedTestException();
}
out.collect(first != null ? first : second);
}
}
}
flink-runtime/src/test/java/org/apache/flink/runtime/operators/FullOuterJoinTaskExternalITCase.java 0 → 100644
浏览文件 @ f3dee23b
/*
* 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;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.java.tuple.Tuple2;
public class FullOuterJoinTaskExternalITCase extends AbstractOuterJoinTaskExternalITCase {
public FullOuterJoinTaskExternalITCase(ExecutionConfig config) {
super(config, DriverStrategy.FULL_OUTER_MERGE);
}
@Override
protected int calculateExpectedCount(int keyCnt1, int valCnt1, int keyCnt2, int valCnt2) {
return valCnt1 * valCnt2 * Math.min(keyCnt1, keyCnt2) + (keyCnt2 > keyCnt1 ? (keyCnt2 - keyCnt1) * valCnt2 : (keyCnt1 - keyCnt2) * valCnt1);
}
@Override
protected AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> getOuterJoinDriver() {
return new FullOuterJoinDriver<>();
}
}
flink-runtime/src/test/java/org/apache/flink/runtime/operators/FullOuterJoinTaskTest.java 0 → 100644
浏览文件 @ f3dee23b
/*
* 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;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.java.tuple.Tuple2;
public class FullOuterJoinTaskTest extends AbstractOuterJoinTaskTest {
public FullOuterJoinTaskTest(ExecutionConfig config) {
super(config, DriverStrategy.FULL_OUTER_MERGE);
}
@Override
protected int calculateExpectedCount(int keyCnt1, int valCnt1, int keyCnt2, int valCnt2) {
return valCnt1 * valCnt2 * Math.min(keyCnt1, keyCnt2) + (keyCnt2 > keyCnt1 ? (keyCnt2 - keyCnt1) * valCnt2 : (keyCnt1 - keyCnt2) * valCnt1);
}
@Override
protected AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> getOuterJoinDriver() {
return new FullOuterJoinDriver<>();
}
}
flink-runtime/src/test/java/org/apache/flink/runtime/operators/LeftOuterJoinTaskExternalITCase.java 0 → 100644
浏览文件 @ f3dee23b
/*
* 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;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.java.tuple.Tuple2;
public class LeftOuterJoinTaskExternalITCase extends AbstractOuterJoinTaskExternalITCase {
public LeftOuterJoinTaskExternalITCase(ExecutionConfig config) {
super(config, DriverStrategy.LEFT_OUTER_MERGE);
}
@Override
protected int calculateExpectedCount(int keyCnt1, int valCnt1, int keyCnt2, int valCnt2) {
return valCnt1 * valCnt2 * Math.min(keyCnt1, keyCnt2) + (keyCnt1 > keyCnt2 ? (keyCnt1 - keyCnt2) * valCnt1 : 0);
}
@Override
protected AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> getOuterJoinDriver() {
return new LeftOuterJoinDriver<>();
}
}
flink-runtime/src/test/java/org/apache/flink/runtime/operators/LeftOuterJoinTaskTest.java 0 → 100644
浏览文件 @ f3dee23b
/*
* 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;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.java.tuple.Tuple2;
public class LeftOuterJoinTaskTest extends AbstractOuterJoinTaskTest {
public LeftOuterJoinTaskTest(ExecutionConfig config) {
super(config, DriverStrategy.LEFT_OUTER_MERGE);
}
@Override
protected int calculateExpectedCount(int keyCnt1, int valCnt1, int keyCnt2, int valCnt2) {
return valCnt1 * valCnt2 * Math.min(keyCnt1, keyCnt2) + (keyCnt1 > keyCnt2 ? (keyCnt1 - keyCnt2) * valCnt1 : 0);
}
@Override
protected AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> getOuterJoinDriver() {
return new LeftOuterJoinDriver<>();
}
}
flink-runtime/src/test/java/org/apache/flink/runtime/operators/RightOuterJoinTaskExternalITCase.java 0 → 100644
浏览文件 @ f3dee23b
/*
* 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;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.java.tuple.Tuple2;
public class RightOuterJoinTaskExternalITCase extends AbstractOuterJoinTaskExternalITCase {
public RightOuterJoinTaskExternalITCase(ExecutionConfig config) {
super(config, DriverStrategy.RIGHT_OUTER_MERGE);
}
@Override
protected int calculateExpectedCount(int keyCnt1, int valCnt1, int keyCnt2, int valCnt2) {
return valCnt1 * valCnt2 * Math.min(keyCnt1, keyCnt2) + (keyCnt2 > keyCnt1 ? (keyCnt2 - keyCnt1) * valCnt2 : 0);
}
@Override
protected AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> getOuterJoinDriver() {
return new RightOuterJoinDriver<>();
}
}
flink-runtime/src/test/java/org/apache/flink/runtime/operators/RightOuterJoinTaskTest.java 0 → 100644
浏览文件 @ f3dee23b
/*
* 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;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.java.tuple.Tuple2;
public class RightOuterJoinTaskTest extends AbstractOuterJoinTaskTest {
public RightOuterJoinTaskTest(ExecutionConfig config) {
super(config, DriverStrategy.RIGHT_OUTER_MERGE);
}
@Override
protected int calculateExpectedCount(int keyCnt1, int valCnt1, int keyCnt2, int valCnt2) {
return valCnt1 * valCnt2 * Math.min(keyCnt1, keyCnt2) + (keyCnt2 > keyCnt1 ? (keyCnt2 - keyCnt1) * valCnt2 : 0);
}
@Override
protected AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> getOuterJoinDriver() {
return new RightOuterJoinDriver<>();
}
}
flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/AbstractSortMergeOuterJoinIteratorITCase.java
浏览文件 @ f3dee23b
......@@ -39,7 +39,12 @@ 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.*;
import org.apache.flink.runtime.operators.testutils.CollectionIterator;
import org.apache.flink.runtime.operators.testutils.DiscardingOutputCollector;
import org.apache.flink.runtime.operators.testutils.DummyInvokable;
import org.apache.flink.runtime.operators.testutils.Match;
import org.apache.flink.runtime.operators.testutils.MatchRemovingJoiner;
import org.apache.flink.runtime.operators.testutils.SimpleTupleJoinFunction;
import org.apache.flink.runtime.operators.testutils.TestData.TupleConstantValueIterator;
import org.apache.flink.runtime.operators.testutils.TestData.TupleGenerator;
import org.apache.flink.runtime.operators.testutils.TestData.TupleGenerator.KeyMode;
......@@ -52,7 +57,13 @@ import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import java.util.*;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
public abstract class AbstractSortMergeOuterJoinIteratorITCase {
......@@ -61,11 +72,6 @@ public abstract class AbstractSortMergeOuterJoinIteratorITCase {
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;
......@@ -76,9 +82,7 @@ public abstract class AbstractSortMergeOuterJoinIteratorITCase {
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;
......@@ -90,14 +94,14 @@ public abstract class AbstractSortMergeOuterJoinIteratorITCase {
public void beforeTest() {
ExecutionConfig config = new ExecutionConfig();
config.disableObjectReuse();
typeInfo1 = TupleTypeInfo.getBasicTupleTypeInfo(String.class, String.class);
typeInfo2 = TupleTypeInfo.getBasicTupleTypeInfo(String.class, Integer.class);
TupleTypeInfo<Tuple2<String, String>> typeInfo1 = TupleTypeInfo.getBasicTupleTypeInfo(String.class, String.class);
TupleTypeInfo<Tuple2<String, Integer>> 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);
pairComp = new GenericPairComparator<>(comparator1, comparator2);
this.memoryManager = new DefaultMemoryManager(MEMORY_SIZE, 1);
this.ioManager = new IOManagerAsync();
......@@ -121,17 +125,18 @@ public abstract class AbstractSortMergeOuterJoinIteratorITCase {
}
}
@SuppressWarnings("unchecked")
protected void testFullOuterWithSample() throws Exception {
CollectionIterator<Tuple2<String, String>> input1 = CollectionIterator.of(
new Tuple2<String, String>("Jack", "Engineering"),
new Tuple2<String, String>("Tim", "Sales"),
new Tuple2<String, String>("Zed", "HR")
new Tuple2<>("Jack", "Engineering"),
new Tuple2<>("Tim", "Sales"),
new Tuple2<>("Zed", "HR")
);
CollectionIterator<Tuple2<String, Integer>> input2 = CollectionIterator.of(
new Tuple2<String, Integer>("Allison", 100),
new Tuple2<String, Integer>("Jack", 200),
new Tuple2<String, Integer>("Zed", 150),
new Tuple2<String, Integer>("Zed", 250)
new Tuple2<>("Allison", 100),
new Tuple2<>("Jack", 200),
new Tuple2<>("Zed", 150),
new Tuple2<>("Zed", 250)
);
OuterJoinType outerJoinType = OuterJoinType.FULL;
......@@ -148,17 +153,18 @@ public abstract class AbstractSortMergeOuterJoinIteratorITCase {
Assert.assertEquals(expected, actual);
}
@SuppressWarnings("unchecked")
protected void testLeftOuterWithSample() throws Exception {
CollectionIterator<Tuple2<String, String>> input1 = CollectionIterator.of(
new Tuple2<String, String>("Jack", "Engineering"),
new Tuple2<String, String>("Tim", "Sales"),
new Tuple2<String, String>("Zed", "HR")
new Tuple2<>("Jack", "Engineering"),
new Tuple2<>("Tim", "Sales"),
new Tuple2<>("Zed", "HR")
);
CollectionIterator<Tuple2<String, Integer>> input2 = CollectionIterator.of(
new Tuple2<String, Integer>("Allison", 100),
new Tuple2<String, Integer>("Jack", 200),
new Tuple2<String, Integer>("Zed", 150),
new Tuple2<String, Integer>("Zed", 250)
new Tuple2<>("Allison", 100),
new Tuple2<>("Jack", 200),
new Tuple2<>("Zed", 150),
new Tuple2<>("Zed", 250)
);
List<Tuple4<String, String, String, Object>> actual = computeOuterJoin(input1, input2, OuterJoinType.LEFT);
......@@ -173,17 +179,18 @@ public abstract class AbstractSortMergeOuterJoinIteratorITCase {
Assert.assertEquals(expected, actual);
}
@SuppressWarnings("unchecked")
protected void testRightOuterWithSample() throws Exception {
CollectionIterator<Tuple2<String, String>> input1 = CollectionIterator.of(
new Tuple2<String, String>("Jack", "Engineering"),
new Tuple2<String, String>("Tim", "Sales"),
new Tuple2<String, String>("Zed", "HR")
new Tuple2<>("Jack", "Engineering"),
new Tuple2<>("Tim", "Sales"),
new Tuple2<>("Zed", "HR")
);
CollectionIterator<Tuple2<String, Integer>> input2 = CollectionIterator.of(
new Tuple2<String, Integer>("Allison", 100),
new Tuple2<String, Integer>("Jack", 200),
new Tuple2<String, Integer>("Zed", 150),
new Tuple2<String, Integer>("Zed", 250)
new Tuple2<>("Allison", 100),
new Tuple2<>("Jack", 200),
new Tuple2<>("Zed", 150),
new Tuple2<>("Zed", 250)
);
List<Tuple4<String, String, String, Object>> actual = computeOuterJoin(input1, input2, OuterJoinType.RIGHT);
......@@ -198,11 +205,12 @@ public abstract class AbstractSortMergeOuterJoinIteratorITCase {
Assert.assertEquals(expected, actual);
}
@SuppressWarnings("unchecked")
protected void testRightSideEmpty() throws Exception {
CollectionIterator<Tuple2<String, String>> input1 = CollectionIterator.of(
new Tuple2<String, String>("Jack", "Engineering"),
new Tuple2<String, String>("Tim", "Sales"),
new Tuple2<String, String>("Zed", "HR")
new Tuple2<>("Jack", "Engineering"),
new Tuple2<>("Tim", "Sales"),
new Tuple2<>("Zed", "HR")
);
CollectionIterator<Tuple2<String, Integer>> input2 = CollectionIterator.of();
......@@ -221,13 +229,14 @@ public abstract class AbstractSortMergeOuterJoinIteratorITCase {
Assert.assertEquals(Collections.<Tuple4<String,String,String,Object>>emptyList(), actualRight);
}
@SuppressWarnings("unchecked")
protected void testLeftSideEmpty() throws Exception {
CollectionIterator<Tuple2<String, String>> input1 = CollectionIterator.of();
CollectionIterator<Tuple2<String, Integer>> input2 = CollectionIterator.of(
new Tuple2<String, Integer>("Allison", 100),
new Tuple2<String, Integer>("Jack", 200),
new Tuple2<String, Integer>("Zed", 150),
new Tuple2<String, Integer>("Zed", 250)
new Tuple2<>("Allison", 100),
new Tuple2<>("Jack", 200),
new Tuple2<>("Zed", 150),
new Tuple2<>("Zed", 250)
);
List<Tuple4<String, String, String, Object>> actualLeft = computeOuterJoin(input1, input2, OuterJoinType.LEFT);
......@@ -246,42 +255,49 @@ public abstract class AbstractSortMergeOuterJoinIteratorITCase {
Assert.assertEquals(expected, actualFull);
}
@SuppressWarnings("unchecked, rawtypes")
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();
AbstractMergeOuterJoinIterator<Tuple2<String, String>, Tuple2<String, Integer>, Tuple4<String, String, String, Object>> iterator =
createOuterJoinIterator(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>>();
ListCollector<Tuple4<String, String, String, Object>> collector = new ListCollector<Tuple4<String, String, String, Object>>(actual);
AbstractMergeOuterJoinIterator iterator =
createOuterJoinIterator(
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<>();
ListCollector<Tuple4<String, String, String, Object>> collector = new ListCollector<>(actual);
while (iterator.callWithNextKey(new SimpleTupleJoinFunction(), collector)) ;
iterator.close();
return actual;
}
@SuppressWarnings("unchecked, rawtypes")
protected void testOuterJoinWithHighNumberOfCommonKeys(OuterJoinType outerJoinType, int input1Size, int input1Duplicates, int input1ValueLength,
float input1KeyDensity, int input2Size, int input2Duplicates, int input2ValueLength, float input2KeyDensity) {
TypeSerializer<Tuple2<Integer, String>> serializer1 = new TupleSerializer<Tuple2<Integer, String>>(
TypeSerializer<Tuple2<Integer, String>> serializer1 = new TupleSerializer<>(
(Class<Tuple2<Integer, String>>) (Class<?>) Tuple2.class,
new TypeSerializer<?>[] { IntSerializer.INSTANCE, StringSerializer.INSTANCE });
TypeSerializer<Tuple2<Integer, String>> serializer2 = new TupleSerializer<Tuple2<Integer, String>>(
new TypeSerializer<?>[]{IntSerializer.INSTANCE, StringSerializer.INSTANCE}
);
TypeSerializer<Tuple2<Integer, String>> serializer2 = new TupleSerializer<>(
(Class<Tuple2<Integer, String>>) (Class<?>) Tuple2.class,
new TypeSerializer<?>[] { IntSerializer.INSTANCE, StringSerializer.INSTANCE });
TypeComparator<Tuple2<Integer, String>> comparator1 = new TupleComparator<Tuple2<Integer, String>>(
new TypeSerializer<?>[]{IntSerializer.INSTANCE, StringSerializer.INSTANCE}
);
TypeComparator<Tuple2<Integer, String>> comparator1 = new TupleComparator<>(
new int[]{0},
new TypeComparator<?>[] { new IntComparator(true) },
new TypeSerializer<?>[] { IntSerializer.INSTANCE });
TypeComparator<Tuple2<Integer, String>> comparator2 = new TupleComparator<Tuple2<Integer, String>>(
new TypeComparator<?>[]{new IntComparator(true)},
new TypeSerializer<?>[]{IntSerializer.INSTANCE}
);
TypeComparator<Tuple2<Integer, String>> comparator2 = new TupleComparator<>(
new int[]{0},
new TypeComparator<?>[] { new IntComparator(true) },
new TypeSerializer<?>[] { IntSerializer.INSTANCE });
new TypeComparator<?>[]{new IntComparator(true)},
new TypeSerializer<?>[]{IntSerializer.INSTANCE}
);
TypePairComparator<Tuple2<Integer, String>, Tuple2<Integer, String>> pairComparator =
new GenericPairComparator<Tuple2<Integer, String>, Tuple2<Integer, String>>(comparator1, comparator2);
TypePairComparator<Tuple2<Integer, String>, Tuple2<Integer, String>> pairComparator = new GenericPairComparator<>(comparator1, comparator2);
this.memoryManager = new DefaultMemoryManager(MEMORY_SIZE, 1);
this.ioManager = new IOManagerAsync();
......@@ -298,16 +314,16 @@ public abstract class AbstractSortMergeOuterJoinIteratorITCase {
final TupleConstantValueIterator const1Iter = new TupleConstantValueIterator(DUPLICATE_KEY, "LEFT String for Duplicate Keys", input1Duplicates);
final TupleConstantValueIterator const2Iter = new TupleConstantValueIterator(DUPLICATE_KEY, "RIGHT String for Duplicate Keys", input2Duplicates);
final List<MutableObjectIterator<Tuple2<Integer, String>>> inList1 = new ArrayList<MutableObjectIterator<Tuple2<Integer, String>>>();
final List<MutableObjectIterator<Tuple2<Integer, String>>> inList1 = new ArrayList<>();
inList1.add(gen1Iter);
inList1.add(const1Iter);
final List<MutableObjectIterator<Tuple2<Integer, String>>> inList2 = new ArrayList<MutableObjectIterator<Tuple2<Integer, String>>>();
final List<MutableObjectIterator<Tuple2<Integer, String>>> inList2 = new ArrayList<>();
inList2.add(gen2Iter);
inList2.add(const2Iter);
MutableObjectIterator<Tuple2<Integer, String>> input1 = new MergeIterator<Tuple2<Integer, String>>(inList1, comparator1.duplicate());
MutableObjectIterator<Tuple2<Integer, String>> input2 = new MergeIterator<Tuple2<Integer, String>>(inList2, comparator2.duplicate());
MutableObjectIterator<Tuple2<Integer, String>> input1 = new MergeIterator<>(inList1, comparator1.duplicate());
MutableObjectIterator<Tuple2<Integer, String>> input2 = new MergeIterator<>(inList2, comparator2.duplicate());
// collect expected data
final Map<Integer, Collection<Match>> expectedMatchesMap = joinValues(
......@@ -333,13 +349,13 @@ public abstract class AbstractSortMergeOuterJoinIteratorITCase {
inList2.add(gen2Iter);
inList2.add(const2Iter);
input1 = new MergeIterator<Tuple2<Integer, String>>(inList1, comparator1.duplicate());
input2 = new MergeIterator<Tuple2<Integer, String>>(inList2, comparator2.duplicate());
input1 = new MergeIterator<>(inList1, comparator1.duplicate());
input2 = new MergeIterator<>(inList2, comparator2.duplicate());
final FlatJoinFunction<Tuple2<Integer, String>, Tuple2<Integer, String>, Tuple2<Integer, String>> joinFunction =
new MatchRemovingJoiner(expectedMatchesMap);
final Collector<Tuple2<Integer, String>> collector = new DiscardingOutputCollector<Tuple2<Integer, String>>();
final Collector<Tuple2<Integer, String>> collector = new DiscardingOutputCollector<>();
// we create this sort-merge iterator with little memory for the block-nested-loops fall-back to make sure it
......@@ -367,7 +383,7 @@ public abstract class AbstractSortMergeOuterJoinIteratorITCase {
}
}
protected abstract <T1, T2> AbstractMergeOuterJoinIterator createOuterJoinIterator(OuterJoinType outerJoinType,
protected abstract <T1, T2, T3> AbstractMergeOuterJoinIterator<T1, T2, T3> createOuterJoinIterator(OuterJoinType outerJoinType,
MutableObjectIterator<T1> input1,
MutableObjectIterator<T2> input2,
TypeSerializer<T1> serializer1, TypeComparator<T1> comparator1,
......@@ -387,7 +403,7 @@ public abstract class AbstractSortMergeOuterJoinIteratorITCase {
Map<Integer, Collection<String>> leftMap,
Map<Integer, Collection<String>> rightMap,
OuterJoinType outerJoinType) {
Map<Integer, Collection<Match>> map = new HashMap<Integer, Collection<Match>>();
Map<Integer, Collection<Match>> map = new HashMap<>();
for (Integer key : leftMap.keySet()) {
Collection<String> leftValues = leftMap.get(key);
......@@ -441,8 +457,8 @@ public abstract class AbstractSortMergeOuterJoinIteratorITCase {
private Map<Integer, Collection<String>> collectData(MutableObjectIterator<Tuple2<Integer, String>> iter)
throws Exception {
final Map<Integer, Collection<String>> map = new HashMap<Integer, Collection<String>>();
Tuple2<Integer, String> pair = new Tuple2<Integer, String>();
final Map<Integer, Collection<String>> map = new HashMap<>();
Tuple2<Integer, String> pair = new Tuple2<>();
while ((pair = iter.next(pair)) != null) {
final Integer key = pair.getField(0);
......
flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/NonReusingSortMergeOuterJoinIteratorITCase.java
浏览文件 @ f3dee23b
......@@ -28,10 +28,11 @@ import org.apache.flink.runtime.operators.sort.AbstractMergeOuterJoinIterator.Ou
import org.apache.flink.util.MutableObjectIterator;
import org.junit.Test;
public class NonReusingSortMergeOuterJoinIteratorITCase extends AbstractSortMergeOuterJoinIteratorITCase {
public class NonReusingSortMergeOuterJoinIteratorITCase extends AbstractSortMergeOuterJoinIteratorITCase {
@Override
protected <T1, T2> AbstractMergeOuterJoinIterator createOuterJoinIterator(OuterJoinType outerJoinType, MutableObjectIterator<T1> input1,
@SuppressWarnings({"unchecked", "rawtypes"})
protected <T1, T2, T3> AbstractMergeOuterJoinIterator createOuterJoinIterator(OuterJoinType outerJoinType, MutableObjectIterator<T1> input1,
MutableObjectIterator<T2> input2, TypeSerializer<T1> serializer1,
TypeComparator<T1> comparator1, TypeSerializer<T2> serializer2, TypeComparator<T2> comparator2,
TypePairComparator<T1, T2> pairComparator, MemoryManager memoryManager, IOManager ioManager,
......
flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/ReusingSortMergeOuterJoinIteratorITCase.java
浏览文件 @ f3dee23b
......@@ -31,7 +31,8 @@ import org.junit.Test;
public class ReusingSortMergeOuterJoinIteratorITCase extends AbstractSortMergeOuterJoinIteratorITCase {
@Override
protected <T1, T2> AbstractMergeOuterJoinIterator createOuterJoinIterator(OuterJoinType outerJoinType, MutableObjectIterator<T1> input1,
@SuppressWarnings({"unchecked", "rawtypes"})
protected <T1, T2, T3> AbstractMergeOuterJoinIterator createOuterJoinIterator(OuterJoinType outerJoinType, MutableObjectIterator<T1> input1,
MutableObjectIterator<T2> input2, TypeSerializer<T1> serializer1,
TypeComparator<T1> comparator1, TypeSerializer<T2> serializer2, TypeComparator<T2> comparator2,
TypePairComparator<T1, T2> pairComparator, MemoryManager memoryManager, IOManager ioManager,
......
flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/BinaryOperatorTestBase.java 0 → 100644
浏览文件 @ f3dee23b
/*
* 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.testutils;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.Function;
import org.apache.flink.api.common.functions.util.FunctionUtils;
import org.apache.flink.api.common.typeutils.TypeComparator;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.TypeSerializerFactory;
import org.apache.flink.api.java.typeutils.runtime.RuntimeSerializerFactory;
import org.apache.flink.configuration.Configuration;
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.PactDriver;
import org.apache.flink.runtime.operators.PactTaskContext;
import org.apache.flink.runtime.operators.ResettablePactDriver;
import org.apache.flink.runtime.operators.sort.UnilateralSortMerger;
import org.apache.flink.runtime.operators.util.TaskConfig;
import org.apache.flink.runtime.taskmanager.TaskManagerRuntimeInfo;
import org.apache.flink.util.Collector;
import org.apache.flink.util.MutableObjectIterator;
import org.apache.flink.util.TestLogger;
import org.junit.After;
import org.junit.Assert;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedList;
import java.util.List;
@RunWith(Parameterized.class)
public class BinaryOperatorTestBase<S extends Function, IN, OUT> extends TestLogger implements PactTaskContext<S, OUT> {
protected static final int PAGE_SIZE = 32 * 1024;
private final IOManager ioManager;
private final MemoryManager memManager;
private final List<MutableObjectIterator<IN>> inputs;
private final List<TypeComparator<IN>> comparators;
private final List<UnilateralSortMerger<IN>> sorters;
private final AbstractInvokable owner;
private final TaskConfig taskConfig;
private final TaskManagerRuntimeInfo taskManageInfo;
protected final long perSortMem;
protected final double perSortFractionMem;
private Collector<OUT> output;
protected int numFileHandles;
private S stub;
private PactDriver<S, IN> driver;
private volatile boolean running = true;
private ExecutionConfig executionConfig;
private List<TypeSerializer<IN>> inputSerializers = new ArrayList<>();
protected BinaryOperatorTestBase(ExecutionConfig executionConfig, long memory, int maxNumSorters, long perSortMemory) {
if (memory < 0 || maxNumSorters < 0 || perSortMemory < 0) {
throw new IllegalArgumentException();
}
final long totalMem = Math.max(memory, 0) + (Math.max(maxNumSorters, 0) * perSortMemory);
this.perSortMem = perSortMemory;
this.perSortFractionMem = (double) perSortMemory / totalMem;
this.ioManager = new IOManagerAsync();
this.memManager = totalMem > 0 ? new DefaultMemoryManager(totalMem, 1) : null;
this.inputs = new ArrayList<>();
this.comparators = new ArrayList<>();
this.sorters = new ArrayList<>();
this.owner = new DummyInvokable();
this.taskConfig = new TaskConfig(new Configuration());
this.executionConfig = executionConfig;
this.taskManageInfo = new TaskManagerRuntimeInfo("localhost", new Configuration());
}
@Parameterized.Parameters
public static Collection<Object[]> getConfigurations() throws IOException {
LinkedList<Object[]> configs = new LinkedList<>();
ExecutionConfig withReuse = new ExecutionConfig();
withReuse.enableObjectReuse();
ExecutionConfig withoutReuse = new ExecutionConfig();
withoutReuse.disableObjectReuse();
Object[] a = {withoutReuse};
configs.add(a);
Object[] b = {withReuse};
configs.add(b);
return configs;
}
public void addInput(MutableObjectIterator<IN> input, TypeSerializer<IN> serializer) {
this.inputs.add(input);
this.sorters.add(null);
this.inputSerializers.add(serializer);
}
@SuppressWarnings("unchecked")
public void addInputSorted(MutableObjectIterator<IN> input, TypeSerializer<IN> serializer, TypeComparator<IN> comp) throws Exception {
this.inputSerializers.add(serializer);
UnilateralSortMerger<IN> sorter = new UnilateralSortMerger<>(
this.memManager,
this.ioManager,
input,
this.owner,
new RuntimeSerializerFactory<>(serializer, (Class<IN>) serializer.createInstance().getClass()),
comp,
this.perSortFractionMem,
32,
0.8f
);
this.sorters.add(sorter);
this.inputs.add(null);
}
public void addDriverComparator(TypeComparator<IN> comparator) {
this.comparators.add(comparator);
}
public void setOutput(Collector<OUT> output) {
this.output = output;
}
public void setOutput(List<OUT> output, TypeSerializer<OUT> outSerializer) {
this.output = new ListOutputCollector<>(output, outSerializer);
}
public int getNumFileHandlesForSort() {
return numFileHandles;
}
public void setNumFileHandlesForSort(int numFileHandles) {
this.numFileHandles = numFileHandles;
}
@SuppressWarnings("rawtypes")
public void testDriver(PactDriver driver, Class stubClass) throws Exception {
testDriverInternal(driver, stubClass);
}
@SuppressWarnings({"unchecked", "rawtypes"})
public void testDriverInternal(PactDriver driver, Class stubClass) throws Exception {
this.driver = driver;
driver.setup(this);
this.stub = (S) stubClass.newInstance();
// regular running logic
this.running = true;
boolean stubOpen = false;
try {
// run the data preparation
try {
driver.prepare();
} catch (Throwable t) {
throw new Exception("The data preparation caused an error: " + t.getMessage(), t);
}
// open stub implementation
try {
FunctionUtils.openFunction(this.stub, getTaskConfig().getStubParameters());
stubOpen = true;
} catch (Throwable t) {
throw new Exception("The user defined 'open()' method caused an exception: " + t.getMessage(), t);
}
if (!running) {
return;
}
// run the user code
driver.run();
// close. We close here such that a regular close throwing an exception marks a task as failed.
if (this.running) {
FunctionUtils.closeFunction(this.stub);
stubOpen = false;
}
this.output.close();
} catch (Exception ex) {
// close the input, but do not report any exceptions, since we already have another root cause
if (stubOpen) {
try {
FunctionUtils.closeFunction(this.stub);
} catch (Throwable ignored) {
}
}
// if resettable driver invoke tear down
if (this.driver instanceof ResettablePactDriver) {
final ResettablePactDriver<?, ?> resDriver = (ResettablePactDriver<?, ?>) this.driver;
try {
resDriver.teardown();
} catch (Throwable t) {
throw new Exception("Error while shutting down an iterative operator: " + t.getMessage(), t);
}
}
// drop exception, if the task was canceled
if (this.running) {
throw ex;
}
} finally {
driver.cleanup();
}
}
@SuppressWarnings({"unchecked", "rawtypes"})
public void testResettableDriver(ResettablePactDriver driver, Class stubClass, int iterations) throws Exception {
driver.setup(this);
for (int i = 0; i < iterations; i++) {
if (i == 0) {
driver.initialize();
} else {
driver.reset();
}
testDriver(driver, stubClass);
}
driver.teardown();
}
public void cancel() throws Exception {
this.running = false;
// compensate for races, where cancel is called before the driver is set
// not that this is an artifact of a bad design of this test base, where the setup
// of the basic properties is not separated from the invocation of the execution logic
while (this.driver == null) {
Thread.sleep(200);
}
this.driver.cancel();
}
// --------------------------------------------------------------------------------------------
@Override
public TaskConfig getTaskConfig() {
return this.taskConfig;
}
@Override
public TaskManagerRuntimeInfo getTaskManagerInfo() {
return this.taskManageInfo;
}
@Override
public ExecutionConfig getExecutionConfig() {
return executionConfig;
}
@Override
public ClassLoader getUserCodeClassLoader() {
return getClass().getClassLoader();
}
@Override
public IOManager getIOManager() {
return this.ioManager;
}
@Override
public MemoryManager getMemoryManager() {
return this.memManager;
}
@Override
public <X> MutableObjectIterator<X> getInput(int index) {
MutableObjectIterator<IN> in = this.inputs.get(index);
if (in == null) {
// waiting from sorter
try {
in = this.sorters.get(index).getIterator();
} catch (InterruptedException e) {
throw new RuntimeException("Interrupted");
}
this.inputs.set(index, in);
}
@SuppressWarnings("unchecked")
MutableObjectIterator<X> input = (MutableObjectIterator<X>) this.inputs.get(index);
return input;
}
@Override
@SuppressWarnings("unchecked")
public <X> TypeSerializerFactory<X> getInputSerializer(int index) {
TypeSerializer<X> ser = (TypeSerializer<X>) this.inputSerializers.get(index);
return new RuntimeSerializerFactory<>(ser, (Class<X>) ser.createInstance().getClass());
}
@Override
public <X> TypeComparator<X> getDriverComparator(int index) {
@SuppressWarnings("unchecked")
TypeComparator<X> comparator = (TypeComparator<X>) this.comparators.get(index);
return comparator;
}
@Override
public S getStub() {
return this.stub;
}
@Override
public Collector<OUT> getOutputCollector() {
return this.output;
}
@Override
public AbstractInvokable getOwningNepheleTask() {
return this.owner;
}
@Override
public String formatLogString(String message) {
return "Driver Tester: " + message;
}
// --------------------------------------------------------------------------------------------
@After
public void shutdownAll() throws Exception {
// 1st, shutdown sorters
for (UnilateralSortMerger<?> sorter : this.sorters) {
if (sorter != null) {
sorter.close();
}
}
this.sorters.clear();
// 2nd, shutdown I/O
this.ioManager.shutdown();
Assert.assertTrue("I/O Manager has not properly shut down.", this.ioManager.isProperlyShutDown());
// last, verify all memory is returned and shutdown mem manager
MemoryManager memMan = getMemoryManager();
if (memMan != null) {
Assert.assertTrue("Memory Manager managed memory was not completely freed.", memMan.verifyEmpty());
memMan.shutdown();
}
}
// --------------------------------------------------------------------------------------------
private static final class ListOutputCollector<OUT> implements Collector<OUT> {
private final List<OUT> output;
private final TypeSerializer<OUT> serializer;
public ListOutputCollector(List<OUT> outputList, TypeSerializer<OUT> serializer) {
this.output = outputList;
this.serializer = serializer;
}
@Override
public void collect(OUT record) {
this.output.add(serializer.copy(record));
}
@Override
public void close() {
}
}
public static final class CountingOutputCollector<OUT> implements Collector<OUT> {
private int num;
@Override
public void collect(OUT record) {
this.num++;
}
@Override
public void close() {
}
public int getNumberOfRecords() {
return this.num;
}
}
}
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或