Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[FLINK-4609] Remove redundant check for null in CrossOperator #2490

Closed
wants to merge 1 commit into from
Closed

[FLINK-4609] Remove redundant check for null in CrossOperator #2490

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
23 changes: 7 additions & 16 deletions flink-java/src/main/java/org/apache/flink/api/java/operators/CrossOperator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,9 +20,9 @@

import java.util.Arrays;

import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.Public;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.InvalidProgramException;
import org.apache.flink.api.common.functions.CrossFunction;
import org.apache.flink.api.common.operators.BinaryOperatorInformation;
Expand Down Expand Up @@ -124,21 +124,12 @@ protected org.apache.flink.api.common.operators.base.CrossOperatorBase<I1, I2, O
@Public
public static final class DefaultCross<I1, I2> extends CrossOperator<I1, I2, Tuple2<I1, I2>> {

private final DataSet<I1> input1;
private final DataSet<I2> input2;

public DefaultCross(DataSet<I1> input1, DataSet<I2> input2, CrossHint hint, String defaultName) {

super(input1, input2, new DefaultCrossFunction<I1, I2>(),
new TupleTypeInfo<Tuple2<I1, I2>>(input1.getType(), input2.getType()),
hint, defaultName);

if (input1 == null || input2 == null) {
throw new NullPointerException();
}

this.input1 = input1;
this.input2 = input2;
new TupleTypeInfo<Tuple2<I1, I2>>(
Preconditions.checkNotNull(input1, "input1 is null").getType(),
Preconditions.checkNotNull(input2, "input2 is null").getType()),
hint, defaultName);
}

/**
Expand All @@ -155,9 +146,9 @@ public <R> CrossOperator<I1, I2, R> with(CrossFunction<I1, I2, R> function) {
if (function == null) {
throw new NullPointerException("Cross function must not be null.");
}
TypeInformation<R> returnType = TypeExtractor.getCrossReturnTypes(function, input1.getType(), input2.getType(),
TypeInformation<R> returnType = TypeExtractor.getCrossReturnTypes(function, getInput1().getType(), getInput2().getType(),
super.getDefaultName(), true);
return new CrossOperator<I1, I2, R>(input1, input2, clean(function), returnType,
return new CrossOperator<I1, I2, R>(getInput1(), getInput2(), clean(function), returnType,
getCrossHint(), Utils.getCallLocationName());
}

Expand Down
5 changes: 3 additions & 2 deletions flink-java/src/main/java/org/apache/flink/api/java/operators/TwoInputOperator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import org.apache.flink.annotation.Public;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.util.Preconditions;

/**
* Base class for operations that operates on two input data sets.
Expand All @@ -37,8 +38,8 @@ public abstract class TwoInputOperator<IN1, IN2, OUT, O extends TwoInputOperator


protected TwoInputOperator(DataSet<IN1> input1, DataSet<IN2> input2, TypeInformation<OUT> resultType) {
super(input1.getExecutionEnvironment(), resultType);

super(Preconditions.checkNotNull(input1, "input1 is null").getExecutionEnvironment(), resultType);
Preconditions.checkNotNull(input2, "input2 is null");
DataSet.checkSameExecutionContext(input1, input2);
this.input1 = input1;
this.input2 = input2;
Expand Down