From 40dce9a0ca9ccf862d55c0f48ba765f644f6e5dc Mon Sep 17 00:00:00 2001 From: Fabian Hueske Date: Wed, 13 Jan 2016 00:22:39 +0100 Subject: [PATCH] [Refactor] [DataSet] Refactor key selector translation in DataSet API. Clean up several compiler warnings. --- .../api/java/operators/CoGroupOperator.java | 174 +++++++----------- .../api/java/operators/DistinctOperator.java | 73 ++++---- .../java/operators/GroupCombineOperator.java | 105 +++++------ .../java/operators/GroupReduceOperator.java | 117 +++++------- .../api/java/operators/JoinOperator.java | 114 +++++------- .../apache/flink/api/java/operators/Keys.java | 148 ++++++++++++--- .../api/java/operators/PartitionOperator.java | 79 ++++---- .../api/java/operators/ReduceOperator.java | 61 +++--- .../api/java/operators/SortedGrouping.java | 11 ++ 9 files changed, 422 insertions(+), 460 deletions(-) diff --git a/flink-java/src/main/java/org/apache/flink/api/java/operators/CoGroupOperator.java b/flink-java/src/main/java/org/apache/flink/api/java/operators/CoGroupOperator.java index 16c2bf6bdab4ea..ca41fc5088c28b 100644 --- a/flink-java/src/main/java/org/apache/flink/api/java/operators/CoGroupOperator.java +++ b/flink-java/src/main/java/org/apache/flink/api/java/operators/CoGroupOperator.java @@ -27,16 +27,13 @@ import org.apache.commons.lang3.tuple.Pair; import org.apache.flink.api.common.InvalidProgramException; import org.apache.flink.api.common.functions.CoGroupFunction; -import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.common.functions.Partitioner; import org.apache.flink.api.common.operators.BinaryOperatorInformation; import org.apache.flink.api.common.operators.DualInputSemanticProperties; import org.apache.flink.api.common.operators.Operator; import org.apache.flink.api.common.operators.Order; import org.apache.flink.api.common.operators.Ordering; -import org.apache.flink.api.common.operators.UnaryOperatorInformation; import org.apache.flink.api.common.operators.base.CoGroupOperatorBase; -import org.apache.flink.api.common.operators.base.MapOperatorBase; import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.api.common.typeutils.CompositeType; import org.apache.flink.api.java.DataSet; @@ -46,13 +43,12 @@ import org.apache.flink.api.java.functions.KeySelector; import org.apache.flink.api.java.operators.Keys.ExpressionKeys; import org.apache.flink.api.java.operators.Keys.IncompatibleKeysException; -import org.apache.flink.api.java.operators.translation.KeyExtractingMapper; +import org.apache.flink.api.java.operators.Keys.SelectorFunctionKeys; import org.apache.flink.api.java.operators.translation.PlanBothUnwrappingCoGroupOperator; import org.apache.flink.api.java.operators.translation.PlanLeftUnwrappingCoGroupOperator; import org.apache.flink.api.java.operators.translation.PlanRightUnwrappingCoGroupOperator; import org.apache.flink.api.java.tuple.Tuple; import org.apache.flink.api.java.tuple.Tuple2; -import org.apache.flink.api.java.typeutils.TupleTypeInfo; import org.apache.flink.api.java.typeutils.TypeExtractor; /** @@ -107,7 +103,7 @@ public CoGroupOperator(DataSet input1, DataSet input2, Keys keys1, K // sanity check solution set key mismatches if (input1 instanceof SolutionSetPlaceHolder) { if (keys1 instanceof ExpressionKeys) { - int[] positions = ((ExpressionKeys) keys1).computeLogicalKeyPositions(); + int[] positions = keys1.computeLogicalKeyPositions(); ((SolutionSetPlaceHolder) input1).checkJoinKeyFields(positions); } else { throw new InvalidProgramException("Currently, the solution set may only be CoGrouped with using tuple field positions."); @@ -115,7 +111,7 @@ public CoGroupOperator(DataSet input1, DataSet input2, Keys keys1, K } if (input2 instanceof SolutionSetPlaceHolder) { if (keys2 instanceof ExpressionKeys) { - int[] positions = ((ExpressionKeys) keys2).computeLogicalKeyPositions(); + int[] positions = keys2.computeLogicalKeyPositions(); ((SolutionSetPlaceHolder) input2).checkJoinKeyFields(positions); } else { throw new InvalidProgramException("Currently, the solution set may only be CoGrouped with using tuple field positions."); @@ -140,15 +136,15 @@ public DualInputSemanticProperties getSemanticProperties() { // offset semantic information by extracted key fields if(props != null && - (this.keys1 instanceof Keys.SelectorFunctionKeys || - this.keys2 instanceof Keys.SelectorFunctionKeys)) { + (this.keys1 instanceof SelectorFunctionKeys || + this.keys2 instanceof SelectorFunctionKeys)) { int numFields1 = this.getInput1Type().getTotalFields(); int numFields2 = this.getInput2Type().getTotalFields(); - int offset1 = (this.keys1 instanceof Keys.SelectorFunctionKeys) ? - ((Keys.SelectorFunctionKeys) this.keys1).getKeyType().getTotalFields() : 0; - int offset2 = (this.keys2 instanceof Keys.SelectorFunctionKeys) ? - ((Keys.SelectorFunctionKeys) this.keys2).getKeyType().getTotalFields() : 0; + int offset1 = (this.keys1 instanceof SelectorFunctionKeys) ? + ((SelectorFunctionKeys) this.keys1).getKeyType().getTotalFields() : 0; + int offset2 = (this.keys2 instanceof SelectorFunctionKeys) ? + ((SelectorFunctionKeys) this.keys2).getKeyType().getTotalFields() : 0; props = SemanticPropUtil.addSourceFieldOffsets(props, numFields1, numFields2, offset1, offset2); } @@ -205,44 +201,44 @@ public Partitioner getPartitioner() { final org.apache.flink.api.common.operators.base.CoGroupOperatorBase po; - if (keys1 instanceof Keys.SelectorFunctionKeys - && keys2 instanceof Keys.SelectorFunctionKeys) { + if (keys1 instanceof SelectorFunctionKeys + && keys2 instanceof SelectorFunctionKeys) { @SuppressWarnings("unchecked") - Keys.SelectorFunctionKeys selectorKeys1 = - (Keys.SelectorFunctionKeys) keys1; + SelectorFunctionKeys selectorKeys1 = + (SelectorFunctionKeys) keys1; @SuppressWarnings("unchecked") - Keys.SelectorFunctionKeys selectorKeys2 = - (Keys.SelectorFunctionKeys) keys2; + SelectorFunctionKeys selectorKeys2 = + (SelectorFunctionKeys) keys2; po = translateSelectorFunctionCoGroup(selectorKeys1, selectorKeys2, function, - getInput1Type(), getInput2Type(), getResultType(), name, input1, input2); + getResultType(), name, input1, input2); po.setParallelism(getParallelism()); po.setCustomPartitioner(customPartitioner); } - else if (keys2 instanceof Keys.SelectorFunctionKeys) { + else if (keys2 instanceof SelectorFunctionKeys) { int[] logicalKeyPositions1 = keys1.computeLogicalKeyPositions(); @SuppressWarnings("unchecked") - Keys.SelectorFunctionKeys selectorKeys2 = (Keys.SelectorFunctionKeys) keys2; + SelectorFunctionKeys selectorKeys2 = (SelectorFunctionKeys) keys2; po = translateSelectorFunctionCoGroupRight(logicalKeyPositions1, selectorKeys2, function, - getInput1Type(), getInput2Type(), getResultType(), name, input1, input2); + getInput1Type(), getResultType(), name, input1, input2); po.setParallelism(getParallelism()); po.setCustomPartitioner(customPartitioner); } - else if (keys1 instanceof Keys.SelectorFunctionKeys) { + else if (keys1 instanceof SelectorFunctionKeys) { @SuppressWarnings("unchecked") - Keys.SelectorFunctionKeys selectorKeys1 = (Keys.SelectorFunctionKeys) keys1; + SelectorFunctionKeys selectorKeys1 = (SelectorFunctionKeys) keys1; int[] logicalKeyPositions2 = keys2.computeLogicalKeyPositions(); po = translateSelectorFunctionCoGroupLeft(selectorKeys1, logicalKeyPositions2, function, - getInput1Type(), getInput2Type(), getResultType(), name, input1, input2); + getInput2Type(), getResultType(), name, input1, input2); } else if ( keys1 instanceof Keys.ExpressionKeys && keys2 instanceof Keys.ExpressionKeys) { @@ -256,8 +252,8 @@ else if ( keys1 instanceof Keys.ExpressionKeys && keys2 instanceof Keys.Expressi int[] logicalKeyPositions2 = keys2.computeLogicalKeyPositions(); CoGroupOperatorBase> op = - new CoGroupOperatorBase>( - function, new BinaryOperatorInformation(getInput1Type(), getInput2Type(), getResultType()), + new CoGroupOperatorBase<>( + function, new BinaryOperatorInformation<>(getInput1Type(), getInput2Type(), getResultType()), logicalKeyPositions1, logicalKeyPositions2, name); op.setFirstInput(input1); @@ -292,44 +288,35 @@ else if ( keys1 instanceof Keys.ExpressionKeys && keys2 instanceof Keys.Expressi private static PlanBothUnwrappingCoGroupOperator translateSelectorFunctionCoGroup( - Keys.SelectorFunctionKeys rawKeys1, Keys.SelectorFunctionKeys rawKeys2, + SelectorFunctionKeys rawKeys1, SelectorFunctionKeys rawKeys2, CoGroupFunction function, - TypeInformation inputType1, TypeInformation inputType2, TypeInformation outputType, String name, + TypeInformation outputType, String name, Operator input1, Operator input2) { @SuppressWarnings("unchecked") - final Keys.SelectorFunctionKeys keys1 = (Keys.SelectorFunctionKeys) rawKeys1; + final SelectorFunctionKeys keys1 = (SelectorFunctionKeys) rawKeys1; @SuppressWarnings("unchecked") - final Keys.SelectorFunctionKeys keys2 = (Keys.SelectorFunctionKeys) rawKeys2; + final SelectorFunctionKeys keys2 = (SelectorFunctionKeys) rawKeys2; - final TypeInformation> typeInfoWithKey1 = new TupleTypeInfo>(keys1.getKeyType(), inputType1); - final TypeInformation> typeInfoWithKey2 = new TupleTypeInfo>(keys2.getKeyType(), inputType2); + final TypeInformation> typeInfoWithKey1 = SelectorFunctionKeys.createTypeWithKey(keys1); + final TypeInformation> typeInfoWithKey2 = SelectorFunctionKeys.createTypeWithKey(keys2); - final KeyExtractingMapper extractor1 = new KeyExtractingMapper(keys1.getKeyExtractor()); - final KeyExtractingMapper extractor2 = new KeyExtractingMapper(keys2.getKeyExtractor()); - - final MapOperatorBase, MapFunction>> keyMapper1 = - new MapOperatorBase, MapFunction>>(extractor1, new UnaryOperatorInformation>(inputType1, typeInfoWithKey1), "Key Extractor 1"); - final MapOperatorBase, MapFunction>> keyMapper2 = - new MapOperatorBase, MapFunction>>(extractor2, new UnaryOperatorInformation>(inputType2, typeInfoWithKey2), "Key Extractor 2"); - final PlanBothUnwrappingCoGroupOperator cogroup = new PlanBothUnwrappingCoGroupOperator(function, keys1, keys2, name, outputType, typeInfoWithKey1, typeInfoWithKey2); + final Operator> keyedInput1 = SelectorFunctionKeys.appendKeyExtractor(input1, keys1); + final Operator> keyedInput2 = SelectorFunctionKeys.appendKeyExtractor(input2, keys2); - cogroup.setFirstInput(keyMapper1); - cogroup.setSecondInput(keyMapper2); + final PlanBothUnwrappingCoGroupOperator cogroup = + new PlanBothUnwrappingCoGroupOperator<>(function, keys1, keys2, name, outputType, typeInfoWithKey1, typeInfoWithKey2); - keyMapper1.setInput(input1); - keyMapper2.setInput(input2); - // set parallelism - keyMapper1.setParallelism(input1.getParallelism()); - keyMapper2.setParallelism(input2.getParallelism()); + cogroup.setFirstInput(keyedInput1); + cogroup.setSecondInput(keyedInput2); return cogroup; } private static PlanRightUnwrappingCoGroupOperator translateSelectorFunctionCoGroupRight( - int[] logicalKeyPositions1, Keys.SelectorFunctionKeys rawKeys2, + int[] logicalKeyPositions1, SelectorFunctionKeys rawKeys2, CoGroupFunction function, - TypeInformation inputType1, TypeInformation inputType2, TypeInformation outputType, String name, + TypeInformation inputType1, TypeInformation outputType, String name, Operator input1, Operator input2) { if(!inputType1.isTupleType()) { @@ -337,22 +324,12 @@ private static PlanRightUnwrappingCoGroupOperator keys2 = (Keys.SelectorFunctionKeys) rawKeys2; - - final TypeInformation> typeInfoWithKey2 = - new TupleTypeInfo>(keys2.getKeyType(), inputType2); - - final KeyExtractingMapper extractor2 = - new KeyExtractingMapper(keys2.getKeyExtractor()); - - final MapOperatorBase, MapFunction>> keyMapper2 = - new MapOperatorBase, MapFunction>>( - extractor2, - new UnaryOperatorInformation>(inputType2, typeInfoWithKey2), - "Key Extractor 2"); + final SelectorFunctionKeys keys2 = (SelectorFunctionKeys) rawKeys2; + final TypeInformation> typeInfoWithKey2 = SelectorFunctionKeys.createTypeWithKey(keys2); + final Operator> keyedInput2 = SelectorFunctionKeys.appendKeyExtractor(input2, keys2); final PlanRightUnwrappingCoGroupOperator cogroup = - new PlanRightUnwrappingCoGroupOperator( + new PlanRightUnwrappingCoGroupOperator<>( function, logicalKeyPositions1, keys2, @@ -362,19 +339,15 @@ private static PlanRightUnwrappingCoGroupOperator PlanLeftUnwrappingCoGroupOperator translateSelectorFunctionCoGroupLeft( - Keys.SelectorFunctionKeys rawKeys1, int[] logicalKeyPositions2, + SelectorFunctionKeys rawKeys1, int[] logicalKeyPositions2, CoGroupFunction function, - TypeInformation inputType1, TypeInformation inputType2, TypeInformation outputType, String name, + TypeInformation inputType2, TypeInformation outputType, String name, Operator input1, Operator input2) { if(!inputType2.isTupleType()) { @@ -382,21 +355,12 @@ private static PlanLeftUnwrappingCoGroupOperator keys1 = (Keys.SelectorFunctionKeys) rawKeys1; - - final TypeInformation> typeInfoWithKey1 = - new TupleTypeInfo>(keys1.getKeyType(), inputType1); - - final KeyExtractingMapper extractor1 = new KeyExtractingMapper(keys1.getKeyExtractor()); - - final MapOperatorBase, MapFunction>> keyMapper1 = - new MapOperatorBase, MapFunction>>( - extractor1, - new UnaryOperatorInformation>(inputType1, typeInfoWithKey1), - "Key Extractor 1"); + final SelectorFunctionKeys keys1 = (SelectorFunctionKeys) rawKeys1; + final TypeInformation> typeInfoWithKey1 = SelectorFunctionKeys.createTypeWithKey(keys1); + final Operator> keyedInput1 = SelectorFunctionKeys.appendKeyExtractor(input1, keys1); final PlanLeftUnwrappingCoGroupOperator cogroup = - new PlanLeftUnwrappingCoGroupOperator( + new PlanLeftUnwrappingCoGroupOperator<>( function, keys1, logicalKeyPositions2, @@ -405,13 +369,9 @@ private static PlanLeftUnwrappingCoGroupOperator input1, DataSet input2) { * @see DataSet */ public CoGroupOperatorSetsPredicate where(int... fields) { - return new CoGroupOperatorSetsPredicate(new Keys.ExpressionKeys(fields, input1.getType())); + return new CoGroupOperatorSetsPredicate(new Keys.ExpressionKeys<>(fields, input1.getType())); } /** @@ -472,7 +432,7 @@ public CoGroupOperatorSetsPredicate where(int... fields) { * @see DataSet */ public CoGroupOperatorSetsPredicate where(String... fields) { - return new CoGroupOperatorSetsPredicate(new Keys.ExpressionKeys(fields, input1.getType())); + return new CoGroupOperatorSetsPredicate(new Keys.ExpressionKeys<>(fields, input1.getType())); } /** @@ -489,7 +449,7 @@ public CoGroupOperatorSetsPredicate where(String... fields) { */ public CoGroupOperatorSetsPredicate where(KeySelector keyExtractor) { TypeInformation keyType = TypeExtractor.getKeySelectorTypes(keyExtractor, input1.getType()); - return new CoGroupOperatorSetsPredicate(new Keys.SelectorFunctionKeys(keyExtractor, input1.getType(), keyType)); + return new CoGroupOperatorSetsPredicate(new SelectorFunctionKeys<>(keyExtractor, input1.getType(), keyType)); } // ---------------------------------------------------------------------------------------- @@ -527,7 +487,7 @@ private CoGroupOperatorSetsPredicate(Keys keys1) { * Call {@link org.apache.flink.api.java.operators.CoGroupOperator.CoGroupOperatorSets.CoGroupOperatorSetsPredicate.CoGroupOperatorWithoutFunction#with(org.apache.flink.api.common.functions.CoGroupFunction)} to finalize the CoGroup transformation. */ public CoGroupOperatorWithoutFunction equalTo(int... fields) { - return createCoGroupOperator(new Keys.ExpressionKeys(fields, input2.getType())); + return createCoGroupOperator(new Keys.ExpressionKeys<>(fields, input2.getType())); } /** @@ -540,7 +500,7 @@ public CoGroupOperatorWithoutFunction equalTo(int... fields) { * Call {@link org.apache.flink.api.java.operators.CoGroupOperator.CoGroupOperatorSets.CoGroupOperatorSetsPredicate.CoGroupOperatorWithoutFunction#with(org.apache.flink.api.common.functions.CoGroupFunction)} to finalize the CoGroup transformation. */ public CoGroupOperatorWithoutFunction equalTo(String... fields) { - return createCoGroupOperator(new Keys.ExpressionKeys(fields, input2.getType())); + return createCoGroupOperator(new Keys.ExpressionKeys<>(fields, input2.getType())); } /** @@ -554,7 +514,7 @@ public CoGroupOperatorWithoutFunction equalTo(String... fields) { */ public CoGroupOperatorWithoutFunction equalTo(KeySelector keyExtractor) { TypeInformation keyType = TypeExtractor.getKeySelectorTypes(keyExtractor, input2.getType()); - return createCoGroupOperator(new Keys.SelectorFunctionKeys(keyExtractor, input2.getType(), keyType)); + return createCoGroupOperator(new SelectorFunctionKeys<>(keyExtractor, input2.getType(), keyType)); } /** @@ -601,8 +561,8 @@ private CoGroupOperatorWithoutFunction(Keys keys2) { this.keys2 = keys2; - this.groupSortKeyOrderFirst = new ArrayList>(); - this.groupSortKeyOrderSecond = new ArrayList>(); + this.groupSortKeyOrderFirst = new ArrayList<>(); + this.groupSortKeyOrderSecond = new ArrayList<>(); } /** @@ -650,7 +610,7 @@ public CoGroupOperator with(CoGroupFunction function) TypeInformation returnType = TypeExtractor.getCoGroupReturnTypes(function, input1.getType(), input2.getType(), Utils.getCallLocationName(), true); - return new CoGroupOperator(input1, input2, keys1, keys2, input1.clean(function), returnType, + return new CoGroupOperator<>(input1, input2, keys1, keys2, input1.clean(function), returnType, groupSortKeyOrderFirst, groupSortKeyOrderSecond, customPartitioner, Utils.getCallLocationName()); } @@ -679,11 +639,11 @@ public CoGroupOperatorWithoutFunction sortFirstGroup(int field, Order order) { if (field >= input1.getType().getArity()) { throw new IllegalArgumentException("Order key out of tuple bounds."); } - ExpressionKeys ek = new ExpressionKeys(new int[]{field}, input1.getType()); + ExpressionKeys ek = new ExpressionKeys<>(new int[]{field}, input1.getType()); int[] groupOrderKeys = ek.computeLogicalKeyPositions(); for (int key : groupOrderKeys) { - this.groupSortKeyOrderFirst.add(new ImmutablePair(key, order)); + this.groupSortKeyOrderFirst.add(new ImmutablePair<>(key, order)); } return this; @@ -709,11 +669,11 @@ public CoGroupOperatorWithoutFunction sortSecondGroup(int field, Order order) { if (field >= input2.getType().getArity()) { throw new IllegalArgumentException("Order key out of tuple bounds."); } - ExpressionKeys ek = new ExpressionKeys(new int[]{field}, input2.getType()); + ExpressionKeys ek = new ExpressionKeys<>(new int[]{field}, input2.getType()); int[] groupOrderKeys = ek.computeLogicalKeyPositions(); for (int key : groupOrderKeys) { - this.groupSortKeyOrderSecond.add(new ImmutablePair(key, order)); + this.groupSortKeyOrderSecond.add(new ImmutablePair<>(key, order)); } return this; @@ -734,11 +694,11 @@ public CoGroupOperatorWithoutFunction sortFirstGroup(String fieldExpression, Ord if (! (input1.getType() instanceof CompositeType)) { throw new InvalidProgramException("Specifying order keys via field positions is only valid for composite data types (pojo / tuple / case class)"); } - ExpressionKeys ek = new ExpressionKeys(new String[]{fieldExpression}, input1.getType()); + ExpressionKeys ek = new ExpressionKeys<>(new String[]{fieldExpression}, input1.getType()); int[] groupOrderKeys = ek.computeLogicalKeyPositions(); for (int key : groupOrderKeys) { - this.groupSortKeyOrderFirst.add(new ImmutablePair(key, order)); + this.groupSortKeyOrderFirst.add(new ImmutablePair<>(key, order)); } return this; @@ -759,11 +719,11 @@ public CoGroupOperatorWithoutFunction sortSecondGroup(String fieldExpression, Or if (! (input2.getType() instanceof CompositeType)) { throw new InvalidProgramException("Specifying order keys via field positions is only valid for composite data types (pojo / tuple / case class)"); } - ExpressionKeys ek = new ExpressionKeys(new String[]{fieldExpression}, input2.getType()); + ExpressionKeys ek = new ExpressionKeys<>(new String[]{fieldExpression}, input2.getType()); int[] groupOrderKeys = ek.computeLogicalKeyPositions(); for (int key : groupOrderKeys) { - this.groupSortKeyOrderSecond.add(new ImmutablePair(key, order)); + this.groupSortKeyOrderSecond.add(new ImmutablePair<>(key, order)); } return this; diff --git a/flink-java/src/main/java/org/apache/flink/api/java/operators/DistinctOperator.java b/flink-java/src/main/java/org/apache/flink/api/java/operators/DistinctOperator.java index ad2335b211f87f..d1d208a46affaf 100644 --- a/flink-java/src/main/java/org/apache/flink/api/java/operators/DistinctOperator.java +++ b/flink-java/src/main/java/org/apache/flink/api/java/operators/DistinctOperator.java @@ -19,31 +19,28 @@ package org.apache.flink.api.java.operators; import org.apache.flink.api.common.functions.GroupReduceFunction; -import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.common.operators.Operator; import org.apache.flink.api.common.operators.SingleInputSemanticProperties; import org.apache.flink.api.common.operators.UnaryOperatorInformation; import org.apache.flink.api.common.operators.base.GroupReduceOperatorBase; -import org.apache.flink.api.common.operators.base.MapOperatorBase; import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.api.common.functions.RichGroupReduceFunction; -import org.apache.flink.api.java.operators.translation.KeyExtractingMapper; +import org.apache.flink.api.java.operators.Keys.SelectorFunctionKeys; import org.apache.flink.api.java.operators.translation.PlanUnwrappingReduceGroupOperator; import org.apache.flink.api.java.tuple.Tuple2; -import org.apache.flink.api.java.typeutils.TupleTypeInfo; import org.apache.flink.util.Collector; import org.apache.flink.api.java.DataSet; /** * This operator represents the application of a "distinct" function on a data set, and the * result data set produced by the function. - * + * * @param The type of the data set made distinct by the operator. */ public class DistinctOperator extends SingleInputOperator> { - + private final Keys keys; - + private final String distinctLocationName; public DistinctOperator(DataSet input, Keys keys, String distinctLocationName) { @@ -53,7 +50,7 @@ public DistinctOperator(DataSet input, Keys keys, String distinctLocationN // if keys is null distinction is done on all fields if (keys == null) { - keys = new Keys.ExpressionKeys(new String[] {Keys.ExpressionKeys.SELECT_ALL_CHAR }, input.getType()); + keys = new Keys.ExpressionKeys<>(new String[] {Keys.ExpressionKeys.SELECT_ALL_CHAR }, input.getType()); } this.keys = keys; @@ -61,79 +58,71 @@ public DistinctOperator(DataSet input, Keys keys, String distinctLocationN @Override protected org.apache.flink.api.common.operators.base.GroupReduceOperatorBase translateToDataFlow(Operator input) { - - final RichGroupReduceFunction function = new DistinctFunction(); + + final RichGroupReduceFunction function = new DistinctFunction<>(); String name = getName() != null ? getName() : "Distinct at " + distinctLocationName; - + if (keys instanceof Keys.ExpressionKeys) { int[] logicalKeyPositions = keys.computeLogicalKeyPositions(); - UnaryOperatorInformation operatorInfo = new UnaryOperatorInformation(getInputType(), getResultType()); + UnaryOperatorInformation operatorInfo = new UnaryOperatorInformation<>(getInputType(), getResultType()); GroupReduceOperatorBase> po = new GroupReduceOperatorBase>(function, operatorInfo, logicalKeyPositions, name); po.setCombinable(true); po.setInput(input); po.setParallelism(getParallelism()); - + // make sure that distinct preserves the partitioning for the fields on which they operate if (getType().isTupleType()) { SingleInputSemanticProperties sProps = new SingleInputSemanticProperties(); - + for (int field : keys.computeLogicalKeyPositions()) { sProps.addForwardedField(field, field); } - + po.setSemanticProperties(sProps); } - - + return po; } - else if (keys instanceof Keys.SelectorFunctionKeys) { - - @SuppressWarnings("unchecked") - Keys.SelectorFunctionKeys selectorKeys = (Keys.SelectorFunctionKeys) keys; + else if (keys instanceof SelectorFunctionKeys) { + @SuppressWarnings("unchecked") + SelectorFunctionKeys selectorKeys = (SelectorFunctionKeys) keys; PlanUnwrappingReduceGroupOperator po = translateSelectorFunctionDistinct( - selectorKeys, function, getInputType(), getResultType(), name, input); - + selectorKeys, function, getResultType(), name, input); + po.setParallelism(this.getParallelism()); - + return po; } else { throw new UnsupportedOperationException("Unrecognized key type."); } } - + // -------------------------------------------------------------------------------------------- - + private static PlanUnwrappingReduceGroupOperator translateSelectorFunctionDistinct( - Keys.SelectorFunctionKeys rawKeys, RichGroupReduceFunction function, - TypeInformation inputType, TypeInformation outputType, String name, Operator input) + SelectorFunctionKeys rawKeys, + RichGroupReduceFunction function, + TypeInformation outputType, + String name, + Operator input) { @SuppressWarnings("unchecked") - final Keys.SelectorFunctionKeys keys = (Keys.SelectorFunctionKeys) rawKeys; + final SelectorFunctionKeys keys = (SelectorFunctionKeys) rawKeys; - TypeInformation> typeInfoWithKey = new TupleTypeInfo>(keys.getKeyType(), inputType); + TypeInformation> typeInfoWithKey = SelectorFunctionKeys.createTypeWithKey(keys); + Operator> keyedInput = SelectorFunctionKeys.appendKeyExtractor(input, keys); - KeyExtractingMapper extractor = new KeyExtractingMapper(keys.getKeyExtractor()); - - PlanUnwrappingReduceGroupOperator reducer = - new PlanUnwrappingReduceGroupOperator(function, keys, name, outputType, typeInfoWithKey, true); - - MapOperatorBase, MapFunction>> mapper = new MapOperatorBase, MapFunction>>(extractor, new UnaryOperatorInformation>(inputType, typeInfoWithKey), "Key Extractor"); + new PlanUnwrappingReduceGroupOperator<>(function, keys, name, outputType, typeInfoWithKey, true); + reducer.setInput(keyedInput); - reducer.setInput(mapper); - mapper.setInput(input); - - // set the mapper's parallelism to the input parallelism to make sure it is chained - mapper.setParallelism(input.getParallelism()); - return reducer; } diff --git a/flink-java/src/main/java/org/apache/flink/api/java/operators/GroupCombineOperator.java b/flink-java/src/main/java/org/apache/flink/api/java/operators/GroupCombineOperator.java index 30cb0be55a76f6..6d02eca3f2afea 100644 --- a/flink-java/src/main/java/org/apache/flink/api/java/operators/GroupCombineOperator.java +++ b/flink-java/src/main/java/org/apache/flink/api/java/operators/GroupCombineOperator.java @@ -19,24 +19,20 @@ package org.apache.flink.api.java.operators; import org.apache.flink.api.common.functions.GroupCombineFunction; -import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.common.operators.Operator; import org.apache.flink.api.common.operators.Order; import org.apache.flink.api.common.operators.Ordering; import org.apache.flink.api.common.operators.SingleInputSemanticProperties; import org.apache.flink.api.common.operators.UnaryOperatorInformation; import org.apache.flink.api.common.operators.base.GroupCombineOperatorBase; -import org.apache.flink.api.common.operators.base.MapOperatorBase; import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.api.java.DataSet; import org.apache.flink.api.java.functions.SemanticPropUtil; -import org.apache.flink.api.java.operators.translation.KeyExtractingMapper; import org.apache.flink.api.java.operators.translation.PlanUnwrappingGroupCombineOperator; import org.apache.flink.api.java.operators.translation.PlanUnwrappingSortedGroupCombineOperator; -import org.apache.flink.api.java.operators.translation.TwoKeyExtractingMapper; +import org.apache.flink.api.java.operators.Keys.SelectorFunctionKeys; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.api.java.tuple.Tuple3; -import org.apache.flink.api.java.typeutils.TupleTypeInfo; /** * This operator behaves like the GroupReduceOperator with Combine but only runs the Combine part which reduces all data @@ -96,9 +92,9 @@ public SingleInputSemanticProperties getSemanticProperties() { // offset semantic information by extracted key fields if(props != null && this.grouper != null && - this.grouper.keys instanceof Keys.SelectorFunctionKeys) { + this.grouper.keys instanceof SelectorFunctionKeys) { - int offset = ((Keys.SelectorFunctionKeys) this.grouper.keys).getKeyType().getTotalFields(); + int offset = ((SelectorFunctionKeys) this.grouper.keys).getKeyType().getTotalFields(); if(this.grouper instanceof SortedGrouping) { offset += ((SortedGrouping) this.grouper).getSortSelectionFunctionKey().getKeyType().getTotalFields(); } @@ -121,9 +117,9 @@ public SingleInputSemanticProperties getSemanticProperties() { // distinguish between grouped reduce and non-grouped reduce if (grouper == null) { // non grouped reduce - UnaryOperatorInformation operatorInfo = new UnaryOperatorInformation(getInputType(), getResultType()); + UnaryOperatorInformation operatorInfo = new UnaryOperatorInformation<>(getInputType(), getResultType()); GroupCombineOperatorBase> po = - new GroupCombineOperatorBase>(function, operatorInfo, new int[0], name); + new GroupCombineOperatorBase<>(function, operatorInfo, new int[0], name); po.setInput(input); // the parallelism for a non grouped reduce can only be 1 @@ -131,33 +127,25 @@ public SingleInputSemanticProperties getSemanticProperties() { return po; } - if (grouper.getKeys() instanceof Keys.SelectorFunctionKeys) { + if (grouper.getKeys() instanceof SelectorFunctionKeys) { @SuppressWarnings("unchecked") - Keys.SelectorFunctionKeys selectorKeys = (Keys.SelectorFunctionKeys) grouper.getKeys(); + SelectorFunctionKeys selectorKeys = (SelectorFunctionKeys) grouper.getKeys(); if (grouper instanceof SortedGrouping) { - SortedGrouping sortedGrouper = (SortedGrouping) grouper; - Keys.SelectorFunctionKeys sortKeys = sortedGrouper.getSortSelectionFunctionKey(); - PlanUnwrappingSortedGroupCombineOperator po = translateSelectorFunctionSortedReducer( - selectorKeys, sortKeys, function, getInputType(), getResultType(), name, input); + SortedGrouping sortedGrouping = (SortedGrouping) grouper; + SelectorFunctionKeys sortKeys = sortedGrouping.getSortSelectionFunctionKey(); + Ordering groupOrder = sortedGrouping.getGroupOrdering(); - // set group order - int[] sortKeyPositions = sortedGrouper.getGroupSortKeyPositions(); - Order[] sortOrders = sortedGrouper.getGroupSortOrders(); - - Ordering o = new Ordering(); - for(int i=0; i < sortKeyPositions.length; i++) { - o.appendOrdering(sortKeyPositions[i], null, sortOrders[i]); - } - po.setGroupOrder(o); + PlanUnwrappingSortedGroupCombineOperator po = + translateSelectorFunctionSortedReducer(selectorKeys, sortKeys, groupOrder, function, getResultType(), name, input); po.setParallelism(this.getParallelism()); return po; } else { PlanUnwrappingGroupCombineOperator po = translateSelectorFunctionReducer( - selectorKeys, function, getInputType(), getResultType(), name, input); + selectorKeys, function, getResultType(), name, input); po.setParallelism(this.getParallelism()); return po; @@ -166,9 +154,9 @@ public SingleInputSemanticProperties getSemanticProperties() { else if (grouper.getKeys() instanceof Keys.ExpressionKeys) { int[] logicalKeyPositions = grouper.getKeys().computeLogicalKeyPositions(); - UnaryOperatorInformation operatorInfo = new UnaryOperatorInformation(getInputType(), getResultType()); + UnaryOperatorInformation operatorInfo = new UnaryOperatorInformation<>(getInputType(), getResultType()); GroupCombineOperatorBase> po = - new GroupCombineOperatorBase>(function, operatorInfo, logicalKeyPositions, name); + new GroupCombineOperatorBase<>(function, operatorInfo, logicalKeyPositions, name); po.setInput(input); po.setParallelism(getParallelism()); @@ -197,53 +185,46 @@ else if (grouper.getKeys() instanceof Keys.ExpressionKeys) { // -------------------------------------------------------------------------------------------- + @SuppressWarnings("unchecked") private static PlanUnwrappingGroupCombineOperator translateSelectorFunctionReducer( - Keys.SelectorFunctionKeys rawKeys, GroupCombineFunction function, - TypeInformation inputType, TypeInformation outputType, String name, Operator input) + SelectorFunctionKeys rawKeys, + GroupCombineFunction function, + TypeInformation outputType, + String name, + Operator input) { - @SuppressWarnings("unchecked") - final Keys.SelectorFunctionKeys keys = (Keys.SelectorFunctionKeys) rawKeys; + final SelectorFunctionKeys keys = (SelectorFunctionKeys) rawKeys; - TypeInformation> typeInfoWithKey = new TupleTypeInfo>(keys.getKeyType(), inputType); + TypeInformation> typeInfoWithKey = SelectorFunctionKeys.createTypeWithKey(keys); + Operator> keyedInput = SelectorFunctionKeys.appendKeyExtractor(input, keys); - KeyExtractingMapper extractor = new KeyExtractingMapper(keys.getKeyExtractor()); - - PlanUnwrappingGroupCombineOperator reducer = new PlanUnwrappingGroupCombineOperator(function, keys, name, outputType, typeInfoWithKey); - - MapOperatorBase, MapFunction>> mapper = new MapOperatorBase, MapFunction>>(extractor, new UnaryOperatorInformation>(inputType, typeInfoWithKey), "Key Extractor"); - - reducer.setInput(mapper); - mapper.setInput(input); - - // set the mapper's parallelism to the input parallelism to make sure it is chained - mapper.setParallelism(input.getParallelism()); + PlanUnwrappingGroupCombineOperator reducer = + new PlanUnwrappingGroupCombineOperator<>(function, keys, name, outputType, typeInfoWithKey); + reducer.setInput(keyedInput); return reducer; } + @SuppressWarnings("unchecked") private static PlanUnwrappingSortedGroupCombineOperator translateSelectorFunctionSortedReducer( - Keys.SelectorFunctionKeys rawGroupingKey, Keys.SelectorFunctionKeys rawSortingKey, GroupCombineFunction function, - TypeInformation inputType, TypeInformation outputType, String name, Operator input) + SelectorFunctionKeys rawGroupingKey, + SelectorFunctionKeys rawSortingKeys, + Ordering groupOrder, + GroupCombineFunction function, + TypeInformation outputType, + String name, + Operator input) { - @SuppressWarnings("unchecked") - final Keys.SelectorFunctionKeys groupingKey = (Keys.SelectorFunctionKeys) rawGroupingKey; - - @SuppressWarnings("unchecked") - final Keys.SelectorFunctionKeys sortingKey = (Keys.SelectorFunctionKeys) rawSortingKey; - - TypeInformation> typeInfoWithKey = new TupleTypeInfo>(groupingKey.getKeyType(), sortingKey.getKeyType(), inputType); - - TwoKeyExtractingMapper extractor = new TwoKeyExtractingMapper(groupingKey.getKeyExtractor(), sortingKey.getKeyExtractor()); - - PlanUnwrappingSortedGroupCombineOperator reducer = new PlanUnwrappingSortedGroupCombineOperator(function, groupingKey, sortingKey, name, outputType, typeInfoWithKey); - - MapOperatorBase, MapFunction>> mapper = new MapOperatorBase, MapFunction>>(extractor, new UnaryOperatorInformation>(inputType, typeInfoWithKey), "Key Extractor"); + final SelectorFunctionKeys groupingKey = (SelectorFunctionKeys) rawGroupingKey; + final SelectorFunctionKeys sortingKey = (SelectorFunctionKeys)rawSortingKeys; + TypeInformation> typeInfoWithKey = SelectorFunctionKeys.createTypeWithKey(groupingKey, sortingKey); - reducer.setInput(mapper); - mapper.setInput(input); + Operator> inputWithKey = SelectorFunctionKeys.appendKeyExtractor(input, groupingKey, sortingKey); - // set the mapper's parallelism to the input parallelism to make sure it is chained - mapper.setParallelism(input.getParallelism()); + PlanUnwrappingSortedGroupCombineOperator reducer = + new PlanUnwrappingSortedGroupCombineOperator<>(function, groupingKey, sortingKey, name, outputType, typeInfoWithKey); + reducer.setInput(inputWithKey); + reducer.setGroupOrder(groupOrder); return reducer; } diff --git a/flink-java/src/main/java/org/apache/flink/api/java/operators/GroupReduceOperator.java b/flink-java/src/main/java/org/apache/flink/api/java/operators/GroupReduceOperator.java index fcbb8886c0e360..5225b33f221680 100644 --- a/flink-java/src/main/java/org/apache/flink/api/java/operators/GroupReduceOperator.java +++ b/flink-java/src/main/java/org/apache/flink/api/java/operators/GroupReduceOperator.java @@ -20,24 +20,20 @@ import org.apache.flink.api.common.functions.GroupCombineFunction; import org.apache.flink.api.common.functions.GroupReduceFunction; -import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.common.operators.Operator; import org.apache.flink.api.common.operators.Order; import org.apache.flink.api.common.operators.Ordering; import org.apache.flink.api.common.operators.SingleInputSemanticProperties; import org.apache.flink.api.common.operators.UnaryOperatorInformation; import org.apache.flink.api.common.operators.base.GroupReduceOperatorBase; -import org.apache.flink.api.common.operators.base.MapOperatorBase; import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.api.common.functions.RichGroupReduceFunction; import org.apache.flink.api.java.functions.SemanticPropUtil; -import org.apache.flink.api.java.operators.translation.KeyExtractingMapper; +import org.apache.flink.api.java.operators.Keys.SelectorFunctionKeys; import org.apache.flink.api.java.operators.translation.PlanUnwrappingReduceGroupOperator; import org.apache.flink.api.java.operators.translation.PlanUnwrappingSortedReduceGroupOperator; -import org.apache.flink.api.java.operators.translation.TwoKeyExtractingMapper; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.api.java.tuple.Tuple3; -import org.apache.flink.api.java.typeutils.TupleTypeInfo; import org.apache.flink.api.java.DataSet; /** @@ -132,9 +128,9 @@ public SingleInputSemanticProperties getSemanticProperties() { // offset semantic information by extracted key fields if(props != null && this.grouper != null && - this.grouper.keys instanceof Keys.SelectorFunctionKeys) { + this.grouper.keys instanceof SelectorFunctionKeys) { - int offset = ((Keys.SelectorFunctionKeys) this.grouper.keys).getKeyType().getTotalFields(); + int offset = ((SelectorFunctionKeys) this.grouper.keys).getKeyType().getTotalFields(); if(this.grouper instanceof SortedGrouping) { offset += ((SortedGrouping) this.grouper).getSortSelectionFunctionKey().getKeyType().getTotalFields(); } @@ -156,9 +152,9 @@ public SingleInputSemanticProperties getSemanticProperties() { // distinguish between grouped reduce and non-grouped reduce if (grouper == null) { // non grouped reduce - UnaryOperatorInformation operatorInfo = new UnaryOperatorInformation(getInputType(), getResultType()); + UnaryOperatorInformation operatorInfo = new UnaryOperatorInformation<>(getInputType(), getResultType()); GroupReduceOperatorBase> po = - new GroupReduceOperatorBase>(function, operatorInfo, new int[0], name); + new GroupReduceOperatorBase<>(function, operatorInfo, new int[0], name); po.setCombinable(combinable); po.setInput(input); @@ -167,34 +163,27 @@ public SingleInputSemanticProperties getSemanticProperties() { return po; } - if (grouper.getKeys() instanceof Keys.SelectorFunctionKeys) { + if (grouper.getKeys() instanceof SelectorFunctionKeys) { @SuppressWarnings("unchecked") - Keys.SelectorFunctionKeys selectorKeys = (Keys.SelectorFunctionKeys) grouper.getKeys(); + SelectorFunctionKeys selectorKeys = (SelectorFunctionKeys) grouper.getKeys(); if (grouper instanceof SortedGrouping) { - SortedGrouping sortedGrouper = (SortedGrouping) grouper; - Keys.SelectorFunctionKeys sortKeys = sortedGrouper.getSortSelectionFunctionKey(); - - PlanUnwrappingSortedReduceGroupOperator po = translateSelectorFunctionSortedReducer( - selectorKeys, sortKeys, function, getInputType(), getResultType(), name, input, isCombinable()); + SortedGrouping sortedGrouping = (SortedGrouping) grouper; + SelectorFunctionKeys sortKeys = sortedGrouping.getSortSelectionFunctionKey(); + Ordering groupOrder = sortedGrouping.getGroupOrdering(); - // set group order - int[] sortKeyPositions = sortedGrouper.getGroupSortKeyPositions(); - Order[] sortOrders = sortedGrouper.getGroupSortOrders(); - - Ordering o = new Ordering(); - for(int i=0; i < sortKeyPositions.length; i++) { - o.appendOrdering(sortKeyPositions[i], null, sortOrders[i]); - } - po.setGroupOrder(o); + PlanUnwrappingSortedReduceGroupOperator po = + translateSelectorFunctionSortedReducer( + selectorKeys, sortKeys, groupOrder, function, getResultType(), name, input, isCombinable() + ); po.setParallelism(this.getParallelism()); po.setCustomPartitioner(grouper.getCustomPartitioner()); return po; } else { PlanUnwrappingReduceGroupOperator po = translateSelectorFunctionReducer( - selectorKeys, function, getInputType(), getResultType(), name, input, isCombinable()); + selectorKeys, function, getResultType(), name, input, isCombinable()); po.setParallelism(this.getParallelism()); po.setCustomPartitioner(grouper.getCustomPartitioner()); @@ -204,9 +193,9 @@ public SingleInputSemanticProperties getSemanticProperties() { else if (grouper.getKeys() instanceof Keys.ExpressionKeys) { int[] logicalKeyPositions = grouper.getKeys().computeLogicalKeyPositions(); - UnaryOperatorInformation operatorInfo = new UnaryOperatorInformation(getInputType(), getResultType()); + UnaryOperatorInformation operatorInfo = new UnaryOperatorInformation<>(getInputType(), getResultType()); GroupReduceOperatorBase> po = - new GroupReduceOperatorBase>(function, operatorInfo, logicalKeyPositions, name); + new GroupReduceOperatorBase<>(function, operatorInfo, logicalKeyPositions, name); po.setCombinable(combinable); po.setInput(input); @@ -216,7 +205,7 @@ else if (grouper.getKeys() instanceof Keys.ExpressionKeys) { // set group order if (grouper instanceof SortedGrouping) { SortedGrouping sortedGrouper = (SortedGrouping) grouper; - + int[] sortKeyPositions = sortedGrouper.getGroupSortKeyPositions(); Order[] sortOrders = sortedGrouper.getGroupSortOrders(); @@ -236,56 +225,50 @@ else if (grouper.getKeys() instanceof Keys.ExpressionKeys) { // -------------------------------------------------------------------------------------------- - + + @SuppressWarnings("unchecked") private static PlanUnwrappingReduceGroupOperator translateSelectorFunctionReducer( - Keys.SelectorFunctionKeys rawKeys, GroupReduceFunction function, - TypeInformation inputType, TypeInformation outputType, String name, Operator input, + SelectorFunctionKeys rawKeys, + GroupReduceFunction function, + TypeInformation outputType, + String name, + Operator input, boolean combinable) { - @SuppressWarnings("unchecked") - final Keys.SelectorFunctionKeys keys = (Keys.SelectorFunctionKeys) rawKeys; - - TypeInformation> typeInfoWithKey = new TupleTypeInfo>(keys.getKeyType(), inputType); - - KeyExtractingMapper extractor = new KeyExtractingMapper(keys.getKeyExtractor()); - - PlanUnwrappingReduceGroupOperator reducer = new PlanUnwrappingReduceGroupOperator(function, keys, name, outputType, typeInfoWithKey, combinable); - - MapOperatorBase, MapFunction>> mapper = new MapOperatorBase, MapFunction>>(extractor, new UnaryOperatorInformation>(inputType, typeInfoWithKey), "Key Extractor"); + SelectorFunctionKeys keys = (SelectorFunctionKeys) rawKeys; + TypeInformation> typeInfoWithKey = SelectorFunctionKeys.createTypeWithKey(keys); + + Operator> keyedInput = SelectorFunctionKeys.appendKeyExtractor(input, keys); + + PlanUnwrappingReduceGroupOperator reducer = + new PlanUnwrappingReduceGroupOperator(function, keys, name, outputType, typeInfoWithKey, combinable); + reducer.setInput(keyedInput); - reducer.setInput(mapper); - mapper.setInput(input); - - // set the mapper's parallelism to the input parallelism to make sure it is chained - mapper.setParallelism(input.getParallelism()); - return reducer; } + @SuppressWarnings("unchecked") private static PlanUnwrappingSortedReduceGroupOperator translateSelectorFunctionSortedReducer( - Keys.SelectorFunctionKeys rawGroupingKey, Keys.SelectorFunctionKeys rawSortingKey, GroupReduceFunction function, - TypeInformation inputType, TypeInformation outputType, String name, Operator input, + SelectorFunctionKeys rawGroupingKey, + SelectorFunctionKeys rawSortingKey, + Ordering groupOrdering, + GroupReduceFunction function, + TypeInformation outputType, + String name, + Operator input, boolean combinable) { - @SuppressWarnings("unchecked") - final Keys.SelectorFunctionKeys groupingKey = (Keys.SelectorFunctionKeys) rawGroupingKey; - - @SuppressWarnings("unchecked") - final Keys.SelectorFunctionKeys sortingKey = (Keys.SelectorFunctionKeys) rawSortingKey; - - TypeInformation> typeInfoWithKey = new TupleTypeInfo>(groupingKey.getKeyType(), sortingKey.getKeyType(), inputType); - - TwoKeyExtractingMapper extractor = new TwoKeyExtractingMapper(groupingKey.getKeyExtractor(), sortingKey.getKeyExtractor()); - - PlanUnwrappingSortedReduceGroupOperator reducer = new PlanUnwrappingSortedReduceGroupOperator(function, groupingKey, sortingKey, name, outputType, typeInfoWithKey, combinable); - - MapOperatorBase, MapFunction>> mapper = new MapOperatorBase, MapFunction>>(extractor, new UnaryOperatorInformation>(inputType, typeInfoWithKey), "Key Extractor"); + final SelectorFunctionKeys groupingKey = (SelectorFunctionKeys) rawGroupingKey; + final SelectorFunctionKeys sortingKey = (SelectorFunctionKeys) rawSortingKey; + TypeInformation> typeInfoWithKey = SelectorFunctionKeys.createTypeWithKey(groupingKey,sortingKey); - reducer.setInput(mapper); - mapper.setInput(input); + Operator> inputWithKey = SelectorFunctionKeys.appendKeyExtractor(input, groupingKey, sortingKey); - // set the mapper's parallelism to the input parallelism to make sure it is chained - mapper.setParallelism(input.getParallelism()); + PlanUnwrappingSortedReduceGroupOperator reducer = + new PlanUnwrappingSortedReduceGroupOperator<>( + function, groupingKey, sortingKey, name, outputType, typeInfoWithKey, combinable); + reducer.setInput(inputWithKey); + reducer.setGroupOrder(groupOrdering); return reducer; } diff --git a/flink-java/src/main/java/org/apache/flink/api/java/operators/JoinOperator.java b/flink-java/src/main/java/org/apache/flink/api/java/operators/JoinOperator.java index fac6b466b3a681..0c0b7109120a0b 100644 --- a/flink-java/src/main/java/org/apache/flink/api/java/operators/JoinOperator.java +++ b/flink-java/src/main/java/org/apache/flink/api/java/operators/JoinOperator.java @@ -25,17 +25,14 @@ import org.apache.flink.api.common.InvalidProgramException; import org.apache.flink.api.common.functions.FlatJoinFunction; import org.apache.flink.api.common.functions.JoinFunction; -import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.common.functions.Partitioner; import org.apache.flink.api.common.functions.RichFlatJoinFunction; import org.apache.flink.api.common.operators.BinaryOperatorInformation; import org.apache.flink.api.common.operators.DualInputSemanticProperties; import org.apache.flink.api.common.operators.Operator; -import org.apache.flink.api.common.operators.UnaryOperatorInformation; import org.apache.flink.api.common.operators.base.JoinOperatorBase; import org.apache.flink.api.common.operators.base.InnerJoinOperatorBase; import org.apache.flink.api.common.operators.base.JoinOperatorBase.JoinHint; -import org.apache.flink.api.common.operators.base.MapOperatorBase; import org.apache.flink.api.common.operators.base.OuterJoinOperatorBase; import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.api.java.DataSet; @@ -47,10 +44,10 @@ import org.apache.flink.api.java.operators.DeltaIteration.SolutionSetPlaceHolder; import org.apache.flink.api.java.operators.Keys.ExpressionKeys; import org.apache.flink.api.java.operators.Keys.IncompatibleKeysException; +import org.apache.flink.api.java.operators.Keys.SelectorFunctionKeys; import org.apache.flink.api.java.operators.join.JoinOperatorSetsBase; import org.apache.flink.api.java.operators.join.JoinType; import org.apache.flink.api.java.operators.join.JoinFunctionAssigner; -import org.apache.flink.api.java.operators.translation.KeyExtractingMapper; import org.apache.flink.api.java.operators.translation.TupleRightUnwrappingJoiner; import org.apache.flink.api.java.operators.translation.TupleLeftUnwrappingJoiner; import org.apache.flink.api.java.operators.translation.TupleUnwrappingJoiner; @@ -104,7 +101,7 @@ protected JoinOperator(DataSet input1, DataSet input2, // sanity check solution set key mismatches if (input1 instanceof SolutionSetPlaceHolder) { if (keys1 instanceof ExpressionKeys) { - int[] positions = ((ExpressionKeys) keys1).computeLogicalKeyPositions(); + int[] positions = keys1.computeLogicalKeyPositions(); ((SolutionSetPlaceHolder) input1).checkJoinKeyFields(positions); } else { throw new InvalidProgramException("Currently, the solution set may only be joined with using tuple field positions."); @@ -112,7 +109,7 @@ protected JoinOperator(DataSet input1, DataSet input2, } if (input2 instanceof SolutionSetPlaceHolder) { if (keys2 instanceof ExpressionKeys) { - int[] positions = ((ExpressionKeys) keys2).computeLogicalKeyPositions(); + int[] positions = keys2.computeLogicalKeyPositions(); ((SolutionSetPlaceHolder) input2).checkJoinKeyFields(positions); } else { throw new InvalidProgramException("Currently, the solution set may only be joined with using tuple field positions."); @@ -260,15 +257,15 @@ public DualInputSemanticProperties getSemanticProperties() { // offset semantic information by extracted key fields if(props != null && - (this.keys1 instanceof Keys.SelectorFunctionKeys || - this.keys2 instanceof Keys.SelectorFunctionKeys)) { + (this.keys1 instanceof SelectorFunctionKeys || + this.keys2 instanceof SelectorFunctionKeys)) { int numFields1 = this.getInput1Type().getTotalFields(); int numFields2 = this.getInput2Type().getTotalFields(); - int offset1 = (this.keys1 instanceof Keys.SelectorFunctionKeys) ? - ((Keys.SelectorFunctionKeys) this.keys1).getKeyType().getTotalFields() : 0; - int offset2 = (this.keys2 instanceof Keys.SelectorFunctionKeys) ? - ((Keys.SelectorFunctionKeys) this.keys2).getKeyType().getTotalFields() : 0; + int offset1 = (this.keys1 instanceof SelectorFunctionKeys) ? + ((SelectorFunctionKeys) this.keys1).getKeyType().getTotalFields() : 0; + int offset2 = (this.keys2 instanceof SelectorFunctionKeys) ? + ((SelectorFunctionKeys) this.keys2).getKeyType().getTotalFields() : 0; props = SemanticPropUtil.addSourceFieldOffsets(props, numFields1, numFields2, offset1, offset2); } @@ -315,40 +312,40 @@ protected boolean udfWithForwardedFieldsSecondAnnotation(Class udfClass) { .withJoinHint(getJoinHint()) .withResultType(getResultType()); - final boolean requiresTupleUnwrapping = keys1 instanceof Keys.SelectorFunctionKeys || keys2 instanceof Keys.SelectorFunctionKeys; + final boolean requiresTupleUnwrapping = keys1 instanceof SelectorFunctionKeys || keys2 instanceof SelectorFunctionKeys; if (requiresTupleUnwrapping) { - if (keys1 instanceof Keys.SelectorFunctionKeys && keys2 instanceof Keys.SelectorFunctionKeys) { + if (keys1 instanceof SelectorFunctionKeys && keys2 instanceof SelectorFunctionKeys) { // Both join sides have a key selector function, so we need to do the // tuple wrapping/unwrapping on both sides. @SuppressWarnings("unchecked") - Keys.SelectorFunctionKeys selectorKeys1 = (Keys.SelectorFunctionKeys) keys1; + SelectorFunctionKeys selectorKeys1 = (SelectorFunctionKeys) keys1; @SuppressWarnings("unchecked") - Keys.SelectorFunctionKeys selectorKeys2 = (Keys.SelectorFunctionKeys) keys2; + SelectorFunctionKeys selectorKeys2 = (SelectorFunctionKeys) keys2; builder = builder .withUdf(new TupleUnwrappingJoiner<>(function)) - .withWrappedInput1(input1, selectorKeys1, getInput1Type()) - .withWrappedInput2(input2, selectorKeys2, getInput2Type()); - } else if (keys2 instanceof Keys.SelectorFunctionKeys) { + .withWrappedInput1(input1, selectorKeys1) + .withWrappedInput2(input2, selectorKeys2); + } else if (keys2 instanceof SelectorFunctionKeys) { // The right side of the join needs the tuple wrapping/unwrapping @SuppressWarnings("unchecked") - Keys.SelectorFunctionKeys selectorKeys2 = (Keys.SelectorFunctionKeys) keys2; + SelectorFunctionKeys selectorKeys2 = (SelectorFunctionKeys) keys2; builder = builder .withUdf(new TupleRightUnwrappingJoiner<>(function)) .withInput1(input1, getInput1Type(), keys1) - .withWrappedInput2(input2, selectorKeys2, getInput2Type()); + .withWrappedInput2(input2, selectorKeys2); } else { // The left side of the join needs the tuple wrapping/unwrapping @SuppressWarnings("unchecked") - Keys.SelectorFunctionKeys selectorKeys1 = (Keys.SelectorFunctionKeys) keys1; + SelectorFunctionKeys selectorKeys1 = (SelectorFunctionKeys) keys1; builder = builder .withUdf(new TupleLeftUnwrappingJoiner<>(function)) - .withWrappedInput1(input1, selectorKeys1, getInput1Type()) + .withWrappedInput1(input1, selectorKeys1) .withInput2(input2, getInput2Type(), keys2); } } else if (keys1 instanceof Keys.ExpressionKeys && keys2 instanceof Keys.ExpressionKeys) { @@ -393,24 +390,24 @@ public JoinOperatorBaseBuilder(String name, JoinType joinType) { public JoinOperatorBaseBuilder withWrappedInput1( Operator input1, - Keys.SelectorFunctionKeys rawKeys1, - TypeInformation inputType1) { - TypeInformation> typeInfoWithKey1 = new TupleTypeInfo<>(rawKeys1.getKeyType(), inputType1); + SelectorFunctionKeys rawKeys1) { - MapOperatorBase, MapFunction>> keyMapper1 = - createKeyMapper(rawKeys1, inputType1, input1, "Key Extractor 1"); + @SuppressWarnings("unchecked") + SelectorFunctionKeys keys1 = (SelectorFunctionKeys)rawKeys1; + TypeInformation> typeInfoWithKey1 = SelectorFunctionKeys.createTypeWithKey(keys1); + Operator> keyMapper1 = SelectorFunctionKeys.appendKeyExtractor(input1, keys1); return this.withInput1(keyMapper1, typeInfoWithKey1, rawKeys1); } public JoinOperatorBaseBuilder withWrappedInput2( Operator input2, - Keys.SelectorFunctionKeys rawKeys2, - TypeInformation inputType2) { - TypeInformation> typeInfoWithKey2 = new TupleTypeInfo<>(rawKeys2.getKeyType(), inputType2); + SelectorFunctionKeys rawKeys2) { - MapOperatorBase, MapFunction>> keyMapper2 = - createKeyMapper(rawKeys2, inputType2, input2, "Key Extractor 2"); + @SuppressWarnings("unchecked") + SelectorFunctionKeys keys2 = (SelectorFunctionKeys)rawKeys2; + TypeInformation> typeInfoWithKey2 = SelectorFunctionKeys.createTypeWithKey(keys2); + Operator> keyMapper2 = SelectorFunctionKeys.appendKeyExtractor(input2, keys2); return withInput2(keyMapper2, typeInfoWithKey2, rawKeys2); } @@ -500,27 +497,6 @@ private OuterJoinOperatorBase.OuterJoinType getOuterJoinType() { throw new UnsupportedOperationException(); } } - - private static MapOperatorBase, MapFunction>> createKeyMapper( - Keys.SelectorFunctionKeys rawKeys, - TypeInformation inputType, - Operator input, - String mapperName) { - - @SuppressWarnings("unchecked") - final Keys.SelectorFunctionKeys keys = (Keys.SelectorFunctionKeys) rawKeys; - final TypeInformation> typeInfoWithKey = new TupleTypeInfo<>(keys.getKeyType(), inputType); - final KeyExtractingMapper extractor = new KeyExtractingMapper<>(keys.getKeyExtractor()); - - final MapOperatorBase, MapFunction>> keyMapper = - new MapOperatorBase, MapFunction>>( - extractor, - new UnaryOperatorInformation<>(inputType, typeInfoWithKey), - mapperName); - keyMapper.setInput(input); - keyMapper.setParallelism(input.getParallelism()); - return keyMapper; - } } } @@ -607,7 +583,7 @@ public void join(IN1 left, IN2 right, Collector out) throws Exception { * @see org.apache.flink.api.java.operators.JoinOperator.ProjectJoin */ public ProjectJoin projectFirst(int... firstFieldIndexes) { - JoinProjection joinProjection = new JoinProjection(getInput1(), getInput2(), getKeys1(), getKeys2(), getJoinHint(), firstFieldIndexes, null); + JoinProjection joinProjection = new JoinProjection<>(getInput1(), getInput2(), getKeys1(), getKeys2(), getJoinHint(), firstFieldIndexes, null); return joinProjection.projectTupleX(); } @@ -633,7 +609,7 @@ public ProjectJoin projectFirst(int... firstFie * @see org.apache.flink.api.java.operators.JoinOperator.ProjectJoin */ public ProjectJoin projectSecond(int... secondFieldIndexes) { - JoinProjection joinProjection = new JoinProjection(getInput1(), getInput2(), getKeys1(), getKeys2(), getJoinHint(), null, secondFieldIndexes); + JoinProjection joinProjection = new JoinProjection<>(getInput1(), getInput2(), getKeys1(), getKeys2(), getJoinHint(), null, secondFieldIndexes); return joinProjection.projectTupleX(); } @@ -901,7 +877,7 @@ public JoinOperatorSetsPredicate where(String... fields) { @Override public JoinOperatorSetsPredicate where(KeySelector keySelector) { TypeInformation keyType = TypeExtractor.getKeySelectorTypes(keySelector, input1.getType()); - return new JoinOperatorSetsPredicate(new Keys.SelectorFunctionKeys<>(keySelector, input1.getType(), keyType)); + return new JoinOperatorSetsPredicate(new SelectorFunctionKeys<>(keySelector, input1.getType(), keyType)); } @@ -965,7 +941,7 @@ public DefaultJoin equalTo(String... fields) { @Override public DefaultJoin equalTo(KeySelector keySelector) { TypeInformation keyType = TypeExtractor.getKeySelectorTypes(keySelector, input2.getType()); - return createDefaultJoin(new Keys.SelectorFunctionKeys<>(keySelector, input2.getType(), keyType)); + return createDefaultJoin(new SelectorFunctionKeys<>(keySelector, input2.getType(), keyType)); } } } @@ -980,7 +956,7 @@ public DefaultJoin equalTo(KeySelector keySelector) { public static final class DefaultFlatJoinFunction extends RichFlatJoinFunction> { private static final long serialVersionUID = 1L; - private final Tuple2 outTuple = new Tuple2(); + private final Tuple2 outTuple = new Tuple2<>(); @Override public void join(T1 first, T2 second, Collector> out) throws Exception { @@ -1071,14 +1047,14 @@ public JoinProjection(DataSet ds1, DataSet ds2, Keys keys1, Keys boolean isSecondTuple; if(ds1.getType() instanceof TupleTypeInfo) { - numFieldsDs1 = ((TupleTypeInfo)ds1.getType()).getArity(); + numFieldsDs1 = ds1.getType().getArity(); isFirstTuple = true; } else { numFieldsDs1 = 1; isFirstTuple = false; } if(ds2.getType() instanceof TupleTypeInfo) { - numFieldsDs2 = ((TupleTypeInfo)ds2.getType()).getArity(); + numFieldsDs2 = ds2.getType().getArity(); isSecondTuple = true; } else { numFieldsDs2 = 1; @@ -1162,12 +1138,8 @@ public JoinProjection(DataSet ds1, DataSet ds2, Keys keys1, Keys protected JoinProjection projectFirst(int... firstFieldIndexes) { boolean isFirstTuple; - - if(ds1.getType() instanceof TupleTypeInfo && firstFieldIndexes.length > 0) { - isFirstTuple = true; - } else { - isFirstTuple = false; - } + + isFirstTuple = ds1.getType() instanceof TupleTypeInfo && firstFieldIndexes.length > 0; if(!isFirstTuple && firstFieldIndexes.length != 0) { // field index provided for non-Tuple input @@ -1226,12 +1198,8 @@ protected JoinProjection projectFirst(int... firstFieldIndexes) { protected JoinProjection projectSecond(int... secondFieldIndexes) { boolean isSecondTuple; - - if(ds2.getType() instanceof TupleTypeInfo && secondFieldIndexes.length > 0) { - isSecondTuple = true; - } else { - isSecondTuple = false; - } + + isSecondTuple = ds2.getType() instanceof TupleTypeInfo && secondFieldIndexes.length > 0; if(!isSecondTuple && secondFieldIndexes.length != 0) { // field index provided for non-Tuple input diff --git a/flink-java/src/main/java/org/apache/flink/api/java/operators/Keys.java b/flink-java/src/main/java/org/apache/flink/api/java/operators/Keys.java index 47c66f4ef7fd48..95ca3006d63db4 100644 --- a/flink-java/src/main/java/org/apache/flink/api/java/operators/Keys.java +++ b/flink-java/src/main/java/org/apache/flink/api/java/operators/Keys.java @@ -26,13 +26,23 @@ import com.google.common.base.Joiner; import org.apache.flink.api.common.InvalidProgramException; +import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.common.functions.Partitioner; +import org.apache.flink.api.common.operators.Operator; +import org.apache.flink.api.common.operators.UnaryOperatorInformation; +import org.apache.flink.api.common.operators.base.MapOperatorBase; import org.apache.flink.api.common.typeinfo.AtomicType; import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.api.common.typeutils.CompositeType; import org.apache.flink.api.common.typeutils.CompositeType.FlatFieldDescriptor; import org.apache.flink.api.java.functions.KeySelector; +import org.apache.flink.api.java.operators.translation.KeyExtractingMapper; +import org.apache.flink.api.java.operators.translation.KeyRemovingMapper; +import org.apache.flink.api.java.operators.translation.TwoKeyExtractingMapper; +import org.apache.flink.api.java.tuple.Tuple2; +import org.apache.flink.api.java.tuple.Tuple3; import org.apache.flink.api.java.typeutils.GenericTypeInfo; +import org.apache.flink.api.java.typeutils.TupleTypeInfo; import org.apache.flink.api.java.typeutils.TupleTypeInfoBase; import org.apache.flink.api.java.typeutils.TypeExtractor; import org.slf4j.Logger; @@ -69,6 +79,7 @@ public boolean isEmpty() { public static class SelectorFunctionKeys extends Keys { private final KeySelector keyExtractor; + private final TypeInformation inputType; private final TypeInformation keyType; private final int[] logicalKeyFields; @@ -81,6 +92,7 @@ public SelectorFunctionKeys(KeySelector keyExtractor, TypeInformation i } this.keyExtractor = keyExtractor; + this.inputType = inputType; this.keyType = keyType; if(!keyType.isKeyType()) { @@ -90,7 +102,7 @@ public SelectorFunctionKeys(KeySelector keyExtractor, TypeInformation i // we have to handle a special case here: // if the keyType is a composite type, we need to select the full type with all its fields. if(keyType instanceof CompositeType) { - ExpressionKeys ek = new ExpressionKeys(new String[] {ExpressionKeys.SELECT_ALL_CHAR}, keyType); + ExpressionKeys ek = new ExpressionKeys<>(new String[]{ExpressionKeys.SELECT_ALL_CHAR}, keyType); logicalKeyFields = ek.computeLogicalKeyPositions(); } else { logicalKeyFields = new int[] {0}; @@ -101,6 +113,10 @@ public TypeInformation getKeyType() { return keyType; } + public TypeInformation getInputType() { + return inputType; + } + public KeySelector getKeyExtractor() { return keyExtractor; } @@ -171,11 +187,92 @@ public void validateCustomPartitioner(Partitioner partitioner, TypeInform } if (typeInfo != null && !(typeInfo instanceof GenericTypeInfo) && (!keyType.equals(typeInfo))) { - throw new InvalidProgramException("The partitioner is imcompatible with the key type. " + throw new InvalidProgramException("The partitioner is incompatible with the key type. " + "Partitioner type: " + typeInfo + " , key type: " + keyType); } } + @SuppressWarnings("unchecked") + public static Operator> appendKeyExtractor( + Operator input, + SelectorFunctionKeys key) + { + + TypeInformation inputType = key.getInputType(); + TypeInformation> typeInfoWithKey = createTypeWithKey(key); + KeyExtractingMapper extractor = new KeyExtractingMapper(key.getKeyExtractor()); + + MapOperatorBase, MapFunction>> mapper = + new MapOperatorBase, MapFunction>>( + extractor, + new UnaryOperatorInformation(inputType, typeInfoWithKey), + "Key Extractor" + ); + + mapper.setInput(input); + mapper.setParallelism(input.getParallelism()); + + return mapper; + } + + @SuppressWarnings("unchecked") + public static Operator> appendKeyExtractor( + Operator input, + SelectorFunctionKeys key1, + SelectorFunctionKeys key2) + { + + TypeInformation inputType = key1.getInputType(); + TypeInformation> typeInfoWithKey = createTypeWithKey(key1, key2); + TwoKeyExtractingMapper extractor = + new TwoKeyExtractingMapper<>(key1.getKeyExtractor(), key2.getKeyExtractor()); + + MapOperatorBase, MapFunction>> mapper = + new MapOperatorBase, MapFunction>>( + extractor, + new UnaryOperatorInformation<>(inputType, typeInfoWithKey), + "Key Extractor" + ); + + mapper.setInput(input); + mapper.setParallelism(input.getParallelism()); + + return mapper; + } + + public static org.apache.flink.api.common.operators.SingleInputOperator appendKeyRemover( + Operator> inputWithKey, + SelectorFunctionKeys key) + { + + TypeInformation inputType = key.getInputType(); + TypeInformation> typeInfoWithKey = createTypeWithKey(key); + + MapOperatorBase, T, MapFunction, T>> mapper = + new MapOperatorBase, T, MapFunction, T>>( + new KeyRemovingMapper(), + new UnaryOperatorInformation<>(typeInfoWithKey, inputType), + "Key Remover" + ); + mapper.setInput(inputWithKey); + mapper.setParallelism(inputWithKey.getParallelism()); + + return mapper; + } + + public static TypeInformation> createTypeWithKey( + SelectorFunctionKeys key) + { + return new TupleTypeInfo<>(key.getKeyType(), key.getInputType()); + } + + public static TypeInformation> createTypeWithKey( + SelectorFunctionKeys key1, + SelectorFunctionKeys key2) + { + return new TupleTypeInfo<>(key1.getKeyType(), key2.getKeyType(), key1.getInputType()); + } + @Override public String toString() { return "Key function (Type: " + keyType + ")"; @@ -228,35 +325,30 @@ public ExpressionKeys(int[] groupingFields, TypeInformation type, boolean all } Preconditions.checkArgument(groupingFields.length > 0, "Grouping fields can not be empty at this point"); - keyFields = new ArrayList(type.getTotalFields()); + keyFields = new ArrayList<>(type.getTotalFields()); // for each key, find the field: - for(int j = 0; j < groupingFields.length; j++) { - int keyPos = groupingFields[j]; - + for (int keyPos : groupingFields) { int offset = 0; - for(int i = 0; i < type.getArity(); i++) { + for (int i = 0; i < type.getArity(); i++) { - TypeInformation fieldType = ((CompositeType) type).getTypeAt(i); - if(i < keyPos) { + TypeInformation fieldType = ((CompositeType) type).getTypeAt(i); + if (i < keyPos) { // not yet there, increment key offset offset += fieldType.getTotalFields(); - } - else { + } else { // arrived at key position if (!fieldType.isKeyType()) { throw new InvalidProgramException("This type (" + fieldType + ") cannot be used as key."); } - if(fieldType instanceof CompositeType) { + if (fieldType instanceof CompositeType) { // add all nested fields of composite type - ((CompositeType) fieldType).getFlatFields("*", offset, keyFields); - } - else if(fieldType instanceof AtomicType) { + ((CompositeType) fieldType).getFlatFields("*", offset, keyFields); + } else if (fieldType instanceof AtomicType) { // add atomic type field keyFields.add(new FlatFieldDescriptor(offset, fieldType)); - } - else { + } else { // type should either be composite or atomic - throw new InvalidProgramException("Field type is neither CompositeType nor AtomicType: "+fieldType); + throw new InvalidProgramException("Field type is neither CompositeType nor AtomicType: " + fieldType); } // go to next key break; @@ -267,7 +359,7 @@ else if(fieldType instanceof AtomicType) { } public static List removeNullElementsFromList(List in) { - List elements = new ArrayList(); + List elements = new ArrayList<>(); for(R e: in) { if(e != null) { elements.add(e); @@ -289,7 +381,7 @@ public ExpressionKeys(String[] expressionsIn, TypeInformation type) { throw new InvalidProgramException("Field expression for atomic type must be equal to '*' or '_'."); } - keyFields = new ArrayList(1); + keyFields = new ArrayList<>(1); keyFields.add(new FlatFieldDescriptor(0, type)); } else { CompositeType cType = (CompositeType) type; @@ -299,9 +391,9 @@ public ExpressionKeys(String[] expressionsIn, TypeInformation type) { LOG.warn("The key expressions contained duplicates. They are now unique"); } // extract the keys on their flat position - keyFields = new ArrayList(expressions.length); - for (int i = 0; i < expressions.length; i++) { - List keys = cType.getFlatFields(expressions[i]); // use separate list to do a size check + keyFields = new ArrayList<>(expressions.length); + for (String expression : expressions) { + List keys = cType.getFlatFields(expression); // use separate list to do a size check for (FlatFieldDescriptor key : keys) { TypeInformation keyType = key.getType(); if (!keyType.isKeyType()) { @@ -311,8 +403,8 @@ public ExpressionKeys(String[] expressionsIn, TypeInformation type) { throw new InvalidProgramException("Field type is neither CompositeType nor AtomicType: " + keyType); } } - if(keys.size() == 0) { - throw new InvalidProgramException("Unable to extract key from expression '"+expressions[i]+"' on key "+cType); + if (keys.size() == 0) { + throw new InvalidProgramException("Unable to extract key from expression '" + expression + "' on key " + cType); } keyFields.addAll(keys); } @@ -351,7 +443,7 @@ public boolean areCompatible(Keys other) throws IncompatibleKeysException { @Override public int[] computeLogicalKeyPositions() { - List logicalKeys = new ArrayList(); + List logicalKeys = new ArrayList<>(); for (FlatFieldDescriptor kd : keyFields) { logicalKeys.add(kd.getPosition()); } @@ -390,7 +482,7 @@ public String toString() { } private static String[] removeDuplicates(String[] in) { - List ret = new LinkedList(); + List ret = new LinkedList<>(); for(String el : in) { if(!ret.contains(el)) { ret.add(el); @@ -406,7 +498,7 @@ private static String[] removeDuplicates(String[] in) { // -------------------------------------------------------------------------------------------- - private static final int[] rangeCheckFields(int[] fields, int maxAllowedField) { + private static int[] rangeCheckFields(int[] fields, int maxAllowedField) { // range check and duplicate eliminate int i = 1, k = 0; diff --git a/flink-java/src/main/java/org/apache/flink/api/java/operators/PartitionOperator.java b/flink-java/src/main/java/org/apache/flink/api/java/operators/PartitionOperator.java index dd9dfb6d9fd2e6..c3d46f20bbdadd 100644 --- a/flink-java/src/main/java/org/apache/flink/api/java/operators/PartitionOperator.java +++ b/flink-java/src/main/java/org/apache/flink/api/java/operators/PartitionOperator.java @@ -20,20 +20,16 @@ import com.google.common.base.Preconditions; -import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.common.functions.Partitioner; import org.apache.flink.api.common.operators.Operator; import org.apache.flink.api.common.operators.UnaryOperatorInformation; -import org.apache.flink.api.common.operators.base.MapOperatorBase; import org.apache.flink.api.common.operators.base.PartitionOperatorBase; import org.apache.flink.api.common.operators.base.PartitionOperatorBase.PartitionMethod; import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.api.common.typeutils.CompositeType; import org.apache.flink.api.java.DataSet; -import org.apache.flink.api.java.operators.translation.KeyExtractingMapper; -import org.apache.flink.api.java.operators.translation.KeyRemovingMapper; +import org.apache.flink.api.java.operators.Keys.SelectorFunctionKeys; import org.apache.flink.api.java.tuple.Tuple2; -import org.apache.flink.api.java.typeutils.TupleTypeInfo; /** * This operator represents a partitioning. @@ -113,34 +109,31 @@ public Partitioner getCustomPartitioner() { // distinguish between partition types if (pMethod == PartitionMethod.REBALANCE) { - UnaryOperatorInformation operatorInfo = new UnaryOperatorInformation(getType(), getType()); - PartitionOperatorBase noop = new PartitionOperatorBase(operatorInfo, pMethod, name); + UnaryOperatorInformation operatorInfo = new UnaryOperatorInformation<>(getType(), getType()); + PartitionOperatorBase rebalancedInput = new PartitionOperatorBase<>(operatorInfo, pMethod, name); + rebalancedInput.setInput(input); + rebalancedInput.setParallelism(getParallelism()); - noop.setInput(input); - noop.setParallelism(getParallelism()); - - return noop; + return rebalancedInput; } else if (pMethod == PartitionMethod.HASH || pMethod == PartitionMethod.CUSTOM || pMethod == PartitionMethod.RANGE) { if (pKeys instanceof Keys.ExpressionKeys) { int[] logicalKeyPositions = pKeys.computeLogicalKeyPositions(); - UnaryOperatorInformation operatorInfo = new UnaryOperatorInformation(getType(), getType()); - PartitionOperatorBase noop = new PartitionOperatorBase(operatorInfo, pMethod, logicalKeyPositions, name); - - noop.setInput(input); - noop.setParallelism(getParallelism()); - noop.setCustomPartitioner(customPartitioner); + UnaryOperatorInformation operatorInfo = new UnaryOperatorInformation<>(getType(), getType()); + PartitionOperatorBase partitionedInput = new PartitionOperatorBase<>(operatorInfo, pMethod, logicalKeyPositions, name); + partitionedInput.setInput(input); + partitionedInput.setParallelism(getParallelism()); + partitionedInput.setCustomPartitioner(customPartitioner); - return noop; + return partitionedInput; } else if (pKeys instanceof Keys.SelectorFunctionKeys) { @SuppressWarnings("unchecked") Keys.SelectorFunctionKeys selectorKeys = (Keys.SelectorFunctionKeys) pKeys; - MapOperatorBase po = translateSelectorFunctionPartitioner(selectorKeys, pMethod, getType(), name, input, getParallelism(), customPartitioner); - return po; + return translateSelectorFunctionPartitioner(selectorKeys, pMethod, name, input, getParallelism(), customPartitioner); } else { throw new UnsupportedOperationException("Unrecognized key type."); @@ -151,34 +144,28 @@ else if (pKeys instanceof Keys.SelectorFunctionKeys) { throw new UnsupportedOperationException("Unsupported partitioning method: " + pMethod.name()); } } - - private static MapOperatorBase, T, ?> translateSelectorFunctionPartitioner(Keys.SelectorFunctionKeys rawKeys, - PartitionMethod pMethod, TypeInformation inputType, String name, Operator input, int partitionDop, Partitioner customPartitioner) + + @SuppressWarnings("unchecked") + private static org.apache.flink.api.common.operators.SingleInputOperator translateSelectorFunctionPartitioner( + SelectorFunctionKeys rawKeys, + PartitionMethod pMethod, + String name, + Operator input, + int partitionDop, + Partitioner customPartitioner) { - @SuppressWarnings("unchecked") - final Keys.SelectorFunctionKeys keys = (Keys.SelectorFunctionKeys) rawKeys; - - TypeInformation> typeInfoWithKey = new TupleTypeInfo>(keys.getKeyType(), inputType); - UnaryOperatorInformation, Tuple2> operatorInfo = new UnaryOperatorInformation, Tuple2>(typeInfoWithKey, typeInfoWithKey); - - KeyExtractingMapper extractor = new KeyExtractingMapper(keys.getKeyExtractor()); - - MapOperatorBase, MapFunction>> keyExtractingMap = new MapOperatorBase, MapFunction>>(extractor, new UnaryOperatorInformation>(inputType, typeInfoWithKey), "Key Extractor"); - PartitionOperatorBase> noop = new PartitionOperatorBase>(operatorInfo, pMethod, new int[]{0}, name); - MapOperatorBase, T, MapFunction, T>> keyRemovingMap = new MapOperatorBase, T, MapFunction, T>>(new KeyRemovingMapper(), new UnaryOperatorInformation, T>(typeInfoWithKey, inputType), "Key Extractor"); + final SelectorFunctionKeys keys = (SelectorFunctionKeys) rawKeys; + TypeInformation> typeInfoWithKey = SelectorFunctionKeys.createTypeWithKey(keys); - keyExtractingMap.setInput(input); - noop.setInput(keyExtractingMap); - keyRemovingMap.setInput(noop); - - noop.setCustomPartitioner(customPartitioner); - - // set parallelism - keyExtractingMap.setParallelism(input.getParallelism()); - noop.setParallelism(partitionDop); - keyRemovingMap.setParallelism(partitionDop); - - return keyRemovingMap; + Operator> keyedInput = SelectorFunctionKeys.appendKeyExtractor(input, keys); + + PartitionOperatorBase> keyedPartitionedInput = + new PartitionOperatorBase<>(new UnaryOperatorInformation<>(typeInfoWithKey, typeInfoWithKey), pMethod, new int[]{0}, name); + keyedPartitionedInput.setInput(keyedInput); + keyedPartitionedInput.setCustomPartitioner(customPartitioner); + keyedPartitionedInput.setParallelism(partitionDop); + + return SelectorFunctionKeys.appendKeyRemover(keyedPartitionedInput, keys); } diff --git a/flink-java/src/main/java/org/apache/flink/api/java/operators/ReduceOperator.java b/flink-java/src/main/java/org/apache/flink/api/java/operators/ReduceOperator.java index 1193da55330e23..6791741e53297a 100644 --- a/flink-java/src/main/java/org/apache/flink/api/java/operators/ReduceOperator.java +++ b/flink-java/src/main/java/org/apache/flink/api/java/operators/ReduceOperator.java @@ -18,20 +18,16 @@ package org.apache.flink.api.java.operators; -import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.common.functions.ReduceFunction; import org.apache.flink.api.common.operators.Operator; import org.apache.flink.api.common.operators.SingleInputSemanticProperties; import org.apache.flink.api.common.operators.UnaryOperatorInformation; -import org.apache.flink.api.common.operators.base.MapOperatorBase; import org.apache.flink.api.common.operators.base.ReduceOperatorBase; import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.api.java.functions.SemanticPropUtil; -import org.apache.flink.api.java.operators.translation.KeyExtractingMapper; -import org.apache.flink.api.java.operators.translation.KeyRemovingMapper; +import org.apache.flink.api.java.operators.Keys.SelectorFunctionKeys; import org.apache.flink.api.java.operators.translation.PlanUnwrappingReduceOperator; import org.apache.flink.api.java.tuple.Tuple2; -import org.apache.flink.api.java.typeutils.TupleTypeInfo; import org.apache.flink.api.java.DataSet; /** @@ -89,9 +85,9 @@ public SingleInputSemanticProperties getSemanticProperties() { // offset semantic information by extracted key fields if(props != null && this.grouper != null && - this.grouper.keys instanceof Keys.SelectorFunctionKeys) { + this.grouper.keys instanceof SelectorFunctionKeys) { - int offset = ((Keys.SelectorFunctionKeys) this.grouper.keys).getKeyType().getTotalFields(); + int offset = ((SelectorFunctionKeys) this.grouper.keys).getKeyType().getTotalFields(); if(this.grouper instanceof SortedGrouping) { offset += ((SortedGrouping) this.grouper).getSortSelectionFunctionKey().getKeyType().getTotalFields(); } @@ -109,9 +105,9 @@ public SingleInputSemanticProperties getSemanticProperties() { // distinguish between grouped reduce and non-grouped reduce if (grouper == null) { // non grouped reduce - UnaryOperatorInformation operatorInfo = new UnaryOperatorInformation(getInputType(), getInputType()); + UnaryOperatorInformation operatorInfo = new UnaryOperatorInformation<>(getInputType(), getInputType()); ReduceOperatorBase> po = - new ReduceOperatorBase>(function, operatorInfo, new int[0], name); + new ReduceOperatorBase<>(function, operatorInfo, new int[0], name); po.setInput(input); // the parallelism for a non grouped reduce can only be 1 @@ -120,13 +116,14 @@ public SingleInputSemanticProperties getSemanticProperties() { return po; } - if (grouper.getKeys() instanceof Keys.SelectorFunctionKeys) { + if (grouper.getKeys() instanceof SelectorFunctionKeys) { // reduce with key selector function @SuppressWarnings("unchecked") - Keys.SelectorFunctionKeys selectorKeys = (Keys.SelectorFunctionKeys) grouper.getKeys(); - - MapOperatorBase po = translateSelectorFunctionReducer(selectorKeys, function, getInputType(), name, input, getParallelism()); + SelectorFunctionKeys selectorKeys = (SelectorFunctionKeys) grouper.getKeys(); + + org.apache.flink.api.common.operators.SingleInputOperator po = + translateSelectorFunctionReducer(selectorKeys, function, getInputType(), name, input, getParallelism()); ((PlanUnwrappingReduceOperator) po.getInput()).setCustomPartitioner(grouper.getCustomPartitioner()); return po; @@ -135,9 +132,9 @@ else if (grouper.getKeys() instanceof Keys.ExpressionKeys) { // reduce with field positions int[] logicalKeyPositions = grouper.getKeys().computeLogicalKeyPositions(); - UnaryOperatorInformation operatorInfo = new UnaryOperatorInformation(getInputType(), getInputType()); + UnaryOperatorInformation operatorInfo = new UnaryOperatorInformation<>(getInputType(), getInputType()); ReduceOperatorBase> po = - new ReduceOperatorBase>(function, operatorInfo, logicalKeyPositions, name); + new ReduceOperatorBase<>(function, operatorInfo, logicalKeyPositions, name); po.setCustomPartitioner(grouper.getCustomPartitioner()); @@ -153,30 +150,24 @@ else if (grouper.getKeys() instanceof Keys.ExpressionKeys) { // -------------------------------------------------------------------------------------------- - private static MapOperatorBase, T, ?> translateSelectorFunctionReducer(Keys.SelectorFunctionKeys rawKeys, - ReduceFunction function, TypeInformation inputType, String name, Operator input, int parallelism) + private static org.apache.flink.api.common.operators.SingleInputOperator translateSelectorFunctionReducer( + SelectorFunctionKeys rawKeys, + ReduceFunction function, + TypeInformation inputType, + String name, + Operator input, + int parallelism) { @SuppressWarnings("unchecked") - final Keys.SelectorFunctionKeys keys = (Keys.SelectorFunctionKeys) rawKeys; - - TypeInformation> typeInfoWithKey = new TupleTypeInfo>(keys.getKeyType(), inputType); - - KeyExtractingMapper extractor = new KeyExtractingMapper(keys.getKeyExtractor()); + final SelectorFunctionKeys keys = (SelectorFunctionKeys) rawKeys; - PlanUnwrappingReduceOperator reducer = new PlanUnwrappingReduceOperator(function, keys, name, inputType, typeInfoWithKey); + TypeInformation> typeInfoWithKey = SelectorFunctionKeys.createTypeWithKey(keys); + Operator> keyedInput = SelectorFunctionKeys.appendKeyExtractor(input, keys); - MapOperatorBase, MapFunction>> keyExtractingMap = new MapOperatorBase, MapFunction>>(extractor, new UnaryOperatorInformation>(inputType, typeInfoWithKey), "Key Extractor"); - MapOperatorBase, T, MapFunction, T>> keyRemovingMap = new MapOperatorBase, T, MapFunction, T>>(new KeyRemovingMapper(), new UnaryOperatorInformation, T>(typeInfoWithKey, inputType), "Key Extractor"); - - keyExtractingMap.setInput(input); - reducer.setInput(keyExtractingMap); - keyRemovingMap.setInput(reducer); - - // set parallelism - keyExtractingMap.setParallelism(input.getParallelism()); + PlanUnwrappingReduceOperator reducer = new PlanUnwrappingReduceOperator<>(function, keys, name, inputType, typeInfoWithKey); + reducer.setInput(keyedInput); reducer.setParallelism(parallelism); - keyRemovingMap.setParallelism(parallelism); - - return keyRemovingMap; + + return SelectorFunctionKeys.appendKeyRemover(reducer, keys); } } diff --git a/flink-java/src/main/java/org/apache/flink/api/java/operators/SortedGrouping.java b/flink-java/src/main/java/org/apache/flink/api/java/operators/SortedGrouping.java index b488dd1bdb36b7..6092d14400e5bd 100644 --- a/flink-java/src/main/java/org/apache/flink/api/java/operators/SortedGrouping.java +++ b/flink-java/src/main/java/org/apache/flink/api/java/operators/SortedGrouping.java @@ -19,6 +19,7 @@ package org.apache.flink.api.java.operators; import org.apache.flink.api.common.functions.GroupCombineFunction; +import org.apache.flink.api.common.operators.Ordering; import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.api.common.typeutils.CompositeType; import org.apache.flink.api.java.Utils; @@ -124,6 +125,16 @@ protected int[] getGroupSortKeyPositions() { protected Order[] getGroupSortOrders() { return this.groupSortOrders; } + + protected Ordering getGroupOrdering() { + + Ordering o = new Ordering(); + for(int i=0; i < this.groupSortKeyPositions.length; i++) { + o.appendOrdering(this.groupSortKeyPositions[i], null, this.groupSortOrders[i]); + } + + return o; + } /** * Uses a custom partitioner for the grouping.