From 130fc0c27c1582db4b9528e1ee7818de8d48bef3 Mon Sep 17 00:00:00 2001 From: Kostas Kloudas Date: Mon, 25 Jan 2016 16:07:39 +0100 Subject: [PATCH] FLINK-3198: Renames and documents better the use of the getDataSet() in Grouping. --- .../api/java/operators/AggregateOperator.java | 6 ++-- .../java/operators/GroupCombineOperator.java | 2 +- .../java/operators/GroupReduceOperator.java | 2 +- .../flink/api/java/operators/Grouping.java | 20 ++++++++--- .../api/java/operators/ReduceOperator.java | 2 +- .../api/java/operators/SortedGrouping.java | 34 +++++++++---------- .../api/java/operators/UnsortedGrouping.java | 26 +++++++------- .../UnionPropertyPropagationTest.java | 2 +- .../operators/ScalaAggregateOperator.java | 6 ++-- .../BatchScalaAPICompletenessTest.scala | 2 +- 10 files changed, 56 insertions(+), 46 deletions(-) diff --git a/flink-java/src/main/java/org/apache/flink/api/java/operators/AggregateOperator.java b/flink-java/src/main/java/org/apache/flink/api/java/operators/AggregateOperator.java index d65ba0d7fba7e..2eaa6a1587bc9 100644 --- a/flink-java/src/main/java/org/apache/flink/api/java/operators/AggregateOperator.java +++ b/flink-java/src/main/java/org/apache/flink/api/java/operators/AggregateOperator.java @@ -95,16 +95,16 @@ public AggregateOperator(DataSet input, Aggregations function, int field, St * @param field */ public AggregateOperator(Grouping input, Aggregations function, int field, String aggregateLocationName) { - super(Preconditions.checkNotNull(input).getDataSet(), input.getDataSet().getType()); + super(Preconditions.checkNotNull(input).getInputDataSet(), input.getInputDataSet().getType()); Preconditions.checkNotNull(function); this.aggregateLocationName = aggregateLocationName; - if (!input.getDataSet().getType().isTupleType()) { + if (!input.getInputDataSet().getType().isTupleType()) { throw new InvalidProgramException("Aggregating on field positions is only possible on tuple data types."); } - TupleTypeInfoBase inType = (TupleTypeInfoBase) input.getDataSet().getType(); + TupleTypeInfoBase inType = (TupleTypeInfoBase) input.getInputDataSet().getType(); if (field < 0 || field >= inType.getArity()) { throw new IllegalArgumentException("Aggregation field position is out of range."); 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 6d02eca3f2afe..ef0c12f74d3b0 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 @@ -72,7 +72,7 @@ public GroupCombineOperator(DataSet input, TypeInformation resultType, * @param function The user-defined GroupReduce function. */ public GroupCombineOperator(Grouping input, TypeInformation resultType, GroupCombineFunction function, String defaultName) { - super(input != null ? input.getDataSet() : null, resultType); + super(input != null ? input.getInputDataSet() : null, resultType); this.function = function; this.grouper = input; 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 5225b33f22168..d501d54a058d6 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 @@ -76,7 +76,7 @@ public GroupReduceOperator(DataSet input, TypeInformation resultType, G * @param function The user-defined GroupReduce function. */ public GroupReduceOperator(Grouping input, TypeInformation resultType, GroupReduceFunction function, String defaultName) { - super(input != null ? input.getDataSet() : null, resultType); + super(input != null ? input.getInputDataSet() : null, resultType); this.function = function; this.grouper = input; diff --git a/flink-java/src/main/java/org/apache/flink/api/java/operators/Grouping.java b/flink-java/src/main/java/org/apache/flink/api/java/operators/Grouping.java index 59811b6537b97..c1174582b923a 100644 --- a/flink-java/src/main/java/org/apache/flink/api/java/operators/Grouping.java +++ b/flink-java/src/main/java/org/apache/flink/api/java/operators/Grouping.java @@ -37,7 +37,7 @@ */ public abstract class Grouping { - protected final DataSet dataSet; + protected final DataSet inputDataSet; protected final Keys keys; @@ -53,13 +53,23 @@ public Grouping(DataSet set, Keys keys) { throw new InvalidProgramException("The grouping keys must not be empty."); } - this.dataSet = set; + this.inputDataSet = set; this.keys = keys; } - - public DataSet getDataSet() { - return this.dataSet; + /** + * Returns the input DataSet of a grouping operation, that is the one before the grouping. This means that + * if it is applied directly to the result of a grouping operation, it will cancel its effect. As an example, in the + * following snippet: + * 

+	 * DataSet notGrouped = input.groupBy().getDataSet();
+	 * DataSet allReduced = notGrouped.reduce()
+	 *
+ * the groupBy() is as if it never happened, as the notGrouped DataSet corresponds + * to the input of the groupBy() (because of the getDataset()). + * */ + public DataSet getInputDataSet() { + return this.inputDataSet; } public Keys getKeys() { 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 6791741e53297..6f8877f1fae9f 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 @@ -63,7 +63,7 @@ public ReduceOperator(DataSet input, ReduceFunction function, String def public ReduceOperator(Grouping input, ReduceFunction function, String defaultName) { - super(input.getDataSet(), input.getDataSet().getType()); + super(input.getInputDataSet(), input.getInputDataSet().getType()); this.function = function; this.grouper = input; 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 6092d14400e5b..e3adefda0dc80 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 @@ -59,16 +59,16 @@ public class SortedGrouping extends Grouping { public SortedGrouping(DataSet set, Keys keys, int field, Order order) { super(set, keys); - if (!dataSet.getType().isTupleType()) { + if (!inputDataSet.getType().isTupleType()) { throw new InvalidProgramException("Specifying order keys via field positions is only valid for tuple data types"); } - if (field >= dataSet.getType().getArity()) { + if (field >= inputDataSet.getType().getArity()) { throw new IllegalArgumentException("Order key out of tuple bounds."); } isValidSortKeyType(field); // use int-based expression key to properly resolve nested tuples for grouping - ExpressionKeys ek = new ExpressionKeys(new int[]{field}, dataSet.getType()); + ExpressionKeys ek = new ExpressionKeys(new int[]{field}, inputDataSet.getType()); this.groupSortKeyPositions = ek.computeLogicalKeyPositions(); this.groupSortOrders = new Order[groupSortKeyPositions.length]; Arrays.fill(this.groupSortOrders, order); @@ -80,13 +80,13 @@ public SortedGrouping(DataSet set, Keys keys, int field, Order order) { public SortedGrouping(DataSet set, Keys keys, String field, Order order) { super(set, keys); - if (!(dataSet.getType() instanceof CompositeType)) { + if (!(inputDataSet.getType() instanceof CompositeType)) { throw new InvalidProgramException("Specifying order keys via field positions is only valid for composite data types (pojo / tuple / case class)"); } isValidSortKeyType(field); // resolve String-field to int using the expression keys - ExpressionKeys ek = new ExpressionKeys(new String[]{field}, dataSet.getType()); + ExpressionKeys ek = new ExpressionKeys(new String[]{field}, inputDataSet.getType()); this.groupSortKeyPositions = ek.computeLogicalKeyPositions(); this.groupSortOrders = new Order[groupSortKeyPositions.length]; Arrays.fill(this.groupSortOrders, order); // if field == "*" @@ -173,8 +173,8 @@ public GroupReduceOperator reduceGroup(GroupReduceFunction reduc throw new NullPointerException("GroupReduce function must not be null."); } TypeInformation resultType = TypeExtractor.getGroupReduceReturnTypes(reducer, - this.getDataSet().getType(), Utils.getCallLocationName(), true); - return new GroupReduceOperator(this, resultType, dataSet.clean(reducer), Utils.getCallLocationName()); + this.getInputDataSet().getType(), Utils.getCallLocationName(), true); + return new GroupReduceOperator(this, resultType, inputDataSet.clean(reducer), Utils.getCallLocationName()); } /** @@ -194,9 +194,9 @@ public GroupCombineOperator combineGroup(GroupCombineFunction co throw new NullPointerException("GroupCombine function must not be null."); } TypeInformation resultType = TypeExtractor.getGroupCombineReturnTypes(combiner, - this.getDataSet().getType(), Utils.getCallLocationName(), true); + this.getInputDataSet().getType(), Utils.getCallLocationName(), true); - return new GroupCombineOperator(this, resultType, dataSet.clean(combiner), Utils.getCallLocationName()); + return new GroupCombineOperator(this, resultType, inputDataSet.clean(combiner), Utils.getCallLocationName()); } @@ -233,15 +233,15 @@ public SortedGrouping sortGroup(int field, Order order) { if (groupSortSelectorFunctionKey != null) { throw new InvalidProgramException("Chaining sortGroup with KeySelector sorting is not supported"); } - if (!dataSet.getType().isTupleType()) { + if (!inputDataSet.getType().isTupleType()) { throw new InvalidProgramException("Specifying order keys via field positions is only valid for tuple data types"); } - if (field >= dataSet.getType().getArity()) { + if (field >= inputDataSet.getType().getArity()) { throw new IllegalArgumentException("Order key out of tuple bounds."); } isValidSortKeyType(field); - ExpressionKeys ek = new ExpressionKeys(new int[]{field}, dataSet.getType()); + ExpressionKeys ek = new ExpressionKeys(new int[]{field}, inputDataSet.getType()); addSortGroupInternal(ek, order); return this; } @@ -262,12 +262,12 @@ public SortedGrouping sortGroup(String field, Order order) { if (groupSortSelectorFunctionKey != null) { throw new InvalidProgramException("Chaining sortGroup with KeySelector sorting is not supported"); } - if (! (dataSet.getType() instanceof CompositeType)) { + if (! (inputDataSet.getType() instanceof CompositeType)) { throw new InvalidProgramException("Specifying order keys via field positions is only valid for composite data types (pojo / tuple / case class)"); } isValidSortKeyType(field); - ExpressionKeys ek = new ExpressionKeys(new String[]{field}, dataSet.getType()); + ExpressionKeys ek = new ExpressionKeys(new String[]{field}, inputDataSet.getType()); addSortGroupInternal(ek, order); return this; } @@ -288,7 +288,7 @@ private void addSortGroupInternal(ExpressionKeys ek, Order order) { } private void isValidSortKeyType(int field) { - TypeInformation sortKeyType = ((TupleTypeInfoBase) dataSet.getType()).getTypeAt(field); + TypeInformation sortKeyType = ((TupleTypeInfoBase) inputDataSet.getType()).getTypeAt(field); if (!sortKeyType.isSortKeyType()) { throw new InvalidProgramException("Selected sort key is not a sortable type " + sortKeyType); } @@ -299,9 +299,9 @@ private void isValidSortKeyType(String field) { field = field.trim(); if(field.equals("*") || field.equals("_")) { - sortKeyType = this.getDataSet().getType(); + sortKeyType = this.getInputDataSet().getType(); } else { - sortKeyType = ((CompositeType) this.getDataSet().getType()).getTypeAt(field); + sortKeyType = ((CompositeType) this.getInputDataSet().getType()).getTypeAt(field); } if (!sortKeyType.isSortKeyType()) { diff --git a/flink-java/src/main/java/org/apache/flink/api/java/operators/UnsortedGrouping.java b/flink-java/src/main/java/org/apache/flink/api/java/operators/UnsortedGrouping.java index 6c2c271f9eacd..5b0a36805a484 100644 --- a/flink-java/src/main/java/org/apache/flink/api/java/operators/UnsortedGrouping.java +++ b/flink-java/src/main/java/org/apache/flink/api/java/operators/UnsortedGrouping.java @@ -136,7 +136,7 @@ public ReduceOperator reduce(ReduceFunction reducer) { if (reducer == null) { throw new NullPointerException("Reduce function must not be null."); } - return new ReduceOperator(this, dataSet.clean(reducer), Utils.getCallLocationName()); + return new ReduceOperator(this, inputDataSet.clean(reducer), Utils.getCallLocationName()); } /** @@ -157,9 +157,9 @@ public GroupReduceOperator reduceGroup(GroupReduceFunction reduc throw new NullPointerException("GroupReduce function must not be null."); } TypeInformation resultType = TypeExtractor.getGroupReduceReturnTypes(reducer, - this.getDataSet().getType(), Utils.getCallLocationName(), true); + this.getInputDataSet().getType(), Utils.getCallLocationName(), true); - return new GroupReduceOperator(this, resultType, dataSet.clean(reducer), Utils.getCallLocationName()); + return new GroupReduceOperator(this, resultType, inputDataSet.clean(reducer), Utils.getCallLocationName()); } /** @@ -179,9 +179,9 @@ public GroupCombineOperator combineGroup(GroupCombineFunction co throw new NullPointerException("GroupCombine function must not be null."); } TypeInformation resultType = TypeExtractor.getGroupCombineReturnTypes(combiner, - this.getDataSet().getType(), Utils.getCallLocationName(), true); + this.getInputDataSet().getType(), Utils.getCallLocationName(), true); - return new GroupCombineOperator(this, resultType, dataSet.clean(combiner), Utils.getCallLocationName()); + return new GroupCombineOperator(this, resultType, inputDataSet.clean(combiner), Utils.getCallLocationName()); } /** @@ -210,12 +210,12 @@ public GroupReduceOperator first(int n) { public ReduceOperator minBy(int... fields) { // Check for using a tuple - if(!this.dataSet.getType().isTupleType()) { + if(!this.inputDataSet.getType().isTupleType()) { throw new InvalidProgramException("Method minBy(int) only works on tuples."); } return new ReduceOperator(this, new SelectByMinFunction( - (TupleTypeInfo) this.dataSet.getType(), fields), Utils.getCallLocationName()); + (TupleTypeInfo) this.inputDataSet.getType(), fields), Utils.getCallLocationName()); } /** @@ -231,12 +231,12 @@ public ReduceOperator minBy(int... fields) { public ReduceOperator maxBy(int... fields) { // Check for using a tuple - if(!this.dataSet.getType().isTupleType()) { + if(!this.inputDataSet.getType().isTupleType()) { throw new InvalidProgramException("Method maxBy(int) only works on tuples."); } return new ReduceOperator(this, new SelectByMaxFunction( - (TupleTypeInfo) this.dataSet.getType(), fields), Utils.getCallLocationName()); + (TupleTypeInfo) this.inputDataSet.getType(), fields), Utils.getCallLocationName()); } // -------------------------------------------------------------------------------------------- // Group Operations @@ -259,7 +259,7 @@ public SortedGrouping sortGroup(int field, Order order) { throw new InvalidProgramException("KeySelector grouping keys and field index group-sorting keys cannot be used together."); } - SortedGrouping sg = new SortedGrouping(this.dataSet, this.keys, field, order); + SortedGrouping sg = new SortedGrouping(this.inputDataSet, this.keys, field, order); sg.customPartitioner = getCustomPartitioner(); return sg; } @@ -280,7 +280,7 @@ public SortedGrouping sortGroup(String field, Order order) { throw new InvalidProgramException("KeySelector grouping keys and field expression group-sorting keys cannot be used together."); } - SortedGrouping sg = new SortedGrouping(this.dataSet, this.keys, field, order); + SortedGrouping sg = new SortedGrouping(this.inputDataSet, this.keys, field, order); sg.customPartitioner = getCustomPartitioner(); return sg; } @@ -301,8 +301,8 @@ public SortedGrouping sortGroup(KeySelector keySelector, Order orde throw new InvalidProgramException("KeySelector group-sorting keys can only be used with KeySelector grouping keys."); } - TypeInformation keyType = TypeExtractor.getKeySelectorTypes(keySelector, this.dataSet.getType()); - SortedGrouping sg = new SortedGrouping(this.dataSet, this.keys, new Keys.SelectorFunctionKeys(keySelector, this.dataSet.getType(), keyType), order); + TypeInformation keyType = TypeExtractor.getKeySelectorTypes(keySelector, this.inputDataSet.getType()); + SortedGrouping sg = new SortedGrouping(this.inputDataSet, this.keys, new Keys.SelectorFunctionKeys(keySelector, this.inputDataSet.getType(), keyType), order); sg.customPartitioner = getCustomPartitioner(); return sg; } diff --git a/flink-optimizer/src/test/java/org/apache/flink/optimizer/UnionPropertyPropagationTest.java b/flink-optimizer/src/test/java/org/apache/flink/optimizer/UnionPropertyPropagationTest.java index c86f30a0e4f66..fefc627b7b749 100644 --- a/flink-optimizer/src/test/java/org/apache/flink/optimizer/UnionPropertyPropagationTest.java +++ b/flink-optimizer/src/test/java/org/apache/flink/optimizer/UnionPropertyPropagationTest.java @@ -98,7 +98,7 @@ public void testUnion2() { final int NUM_INPUTS = 4; // construct the plan it will be multiple flat maps, all unioned - // and the "unioned" dataSet will be grouped + // and the "unioned" inputDataSet will be grouped final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); DataSet source = env.readTextFile(IN_FILE); diff --git a/flink-scala/src/main/java/org/apache/flink/api/scala/operators/ScalaAggregateOperator.java b/flink-scala/src/main/java/org/apache/flink/api/scala/operators/ScalaAggregateOperator.java index ab32ca6291a83..6a14165aafe79 100644 --- a/flink-scala/src/main/java/org/apache/flink/api/scala/operators/ScalaAggregateOperator.java +++ b/flink-scala/src/main/java/org/apache/flink/api/scala/operators/ScalaAggregateOperator.java @@ -95,15 +95,15 @@ public ScalaAggregateOperator(org.apache.flink.api.java.DataSet input, Aggre * @param field */ public ScalaAggregateOperator(Grouping input, Aggregations function, int field) { - super(Preconditions.checkNotNull(input).getDataSet(), input.getDataSet().getType()); + super(Preconditions.checkNotNull(input).getInputDataSet(), input.getInputDataSet().getType()); Preconditions.checkNotNull(function); - if (!input.getDataSet().getType().isTupleType()) { + if (!input.getInputDataSet().getType().isTupleType()) { throw new InvalidProgramException("Aggregating on field positions is only possible on tuple data types."); } - TupleTypeInfoBase inType = (TupleTypeInfoBase) input.getDataSet().getType(); + TupleTypeInfoBase inType = (TupleTypeInfoBase) input.getInputDataSet().getType(); if (field < 0 || field >= inType.getArity()) { throw new IllegalArgumentException("Aggregation field position is out of range."); diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/completeness/BatchScalaAPICompletenessTest.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/completeness/BatchScalaAPICompletenessTest.scala index d50186e562905..36ded0638cc1e 100644 --- a/flink-tests/src/test/scala/org/apache/flink/api/scala/completeness/BatchScalaAPICompletenessTest.scala +++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/completeness/BatchScalaAPICompletenessTest.scala @@ -45,7 +45,7 @@ class BatchScalaAPICompletenessTest extends ScalaAPICompletenessTestBase { "org.apache.flink.api.java.DataSet.getType", "org.apache.flink.api.java.operators.Operator.getResultType", "org.apache.flink.api.java.operators.Operator.getName", - "org.apache.flink.api.java.operators.Grouping.getDataSet", + "org.apache.flink.api.java.operators.Grouping.getInputDataSet", "org.apache.flink.api.java.operators.Grouping.getKeys", "org.apache.flink.api.java.operators.SingleInputOperator.getInput", "org.apache.flink.api.java.operators.SingleInputOperator.getInputType",