You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fh...@apache.org on 2015/12/08 00:57:16 UTC
[2/3] flink git commit: [FLINK-3113] Remove unused and unsupported
global-order methods from GenericDataSinkBase
[FLINK-3113] Remove unused and unsupported global-order methods from GenericDataSinkBase
This closes #1435
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/9547f08a
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/9547f08a
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/9547f08a
Branch: refs/heads/master
Commit: 9547f08aad0fb1e36bb1aa90de17d442b3dc1612
Parents: d144e86
Author: Fabian Hueske <fh...@apache.org>
Authored: Fri Dec 4 13:10:33 2015 +0100
Committer: Fabian Hueske <fh...@apache.org>
Committed: Mon Dec 7 23:11:21 2015 +0100
----------------------------------------------------------------------
.../common/operators/GenericDataSinkBase.java | 78 +-------------------
.../flink/optimizer/dag/DataSinkNode.java | 15 +---
2 files changed, 3 insertions(+), 90 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/9547f08a/flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSinkBase.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSinkBase.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSinkBase.java
index 00cef92..f23a9f2 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSinkBase.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSinkBase.java
@@ -24,7 +24,6 @@ import java.util.Collections;
import java.util.Comparator;
import org.apache.flink.api.common.ExecutionConfig;
-import org.apache.flink.api.common.distributions.DataDistribution;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.io.OutputFormat;
import org.apache.flink.api.common.io.RichOutputFormat;
@@ -53,10 +52,6 @@ public class GenericDataSinkBase<IN> extends Operator<Nothing> {
private Ordering localOrdering;
- private Ordering partitionOrdering;
-
- private DataDistribution distribution;
-
// --------------------------------------------------------------------------------------------
/**
@@ -157,31 +152,6 @@ public class GenericDataSinkBase<IN> extends Operator<Nothing> {
}
// --------------------------------------------------------------------------------------------
-
- /**
- * Sets the order in which the sink must write its data. For any value other then <tt>NONE</tt>,
- * this will cause the system to perform a global sort, or try to reuse an order from a
- * previous operation.
- *
- * @param globalOrder The order to write the data in.
- */
- public void setGlobalOrder(Ordering globalOrder) {
- this.localOrdering = globalOrder;
- setRangePartitioned(globalOrder);
- }
-
- /**
- * Sets the order in which the sink must write its data. For any value other then <tt>NONE</tt>,
- * this will cause the system to perform a global sort, or try to reuse an order from a
- * previous operation.
- *
- * @param globalOrder The order to write the data in.
- * @param distribution The distribution to use for the range partitioning.
- */
- public void setGlobalOrder(Ordering globalOrder, DataDistribution distribution) {
- this.localOrdering = globalOrder;
- setRangePartitioned(globalOrder, distribution);
- }
/**
* Gets the order, in which the data sink writes its data locally. Local order means that
@@ -205,53 +175,7 @@ public class GenericDataSinkBase<IN> extends Operator<Nothing> {
public void setLocalOrder(Ordering localOrder) {
this.localOrdering = localOrder;
}
-
- /**
- * Gets the record ordering over which the sink partitions in ranges.
- *
- * @return The record ordering over which to partition in ranges.
- */
- public Ordering getPartitionOrdering() {
- return this.partitionOrdering;
- }
-
- /**
- * Sets the sink to partition the records into ranges over the given ordering.
- *
- * @param partitionOrdering The record ordering over which to partition in ranges.
- */
- public void setRangePartitioned(Ordering partitionOrdering) {
- throw new UnsupportedOperationException(
- "Range partitioning is currently only supported with a user supplied data distribution.");
- }
-
- /**
- * Sets the sink to partition the records into ranges over the given ordering.
- * The bucket boundaries are determined using the given data distribution.
- *
- * @param partitionOrdering The record ordering over which to partition in ranges.
- * @param distribution The distribution to use for the range partitioning.
- */
- public void setRangePartitioned(Ordering partitionOrdering, DataDistribution distribution) {
- if (partitionOrdering.getNumberOfFields() != distribution.getNumberOfFields()) {
- throw new IllegalArgumentException("The number of keys in the distribution must match number of ordered fields.");
- }
-
- // TODO: check compatibility of distribution and ordering (number and order of keys, key types, etc.
- // TODO: adapt partition ordering to data distribution (use prefix of ordering)
-
- this.partitionOrdering = partitionOrdering;
- this.distribution = distribution;
- }
-
- /**
- * Gets the distribution to use for the range partitioning.
- *
- * @return The distribution to use for the range partitioning.
- */
- public DataDistribution getDataDistribution() {
- return this.distribution;
- }
+
// --------------------------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/9547f08a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/DataSinkNode.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/DataSinkNode.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/DataSinkNode.java
index b35613f..ad50fa2 100644
--- a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/DataSinkNode.java
+++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/DataSinkNode.java
@@ -24,7 +24,6 @@ import java.util.List;
import java.util.Map;
import org.apache.flink.api.common.ExecutionMode;
-import org.apache.flink.api.common.distributions.DataDistribution;
import org.apache.flink.api.common.operators.GenericDataSinkBase;
import org.apache.flink.api.common.operators.Operator;
import org.apache.flink.api.common.operators.Ordering;
@@ -139,22 +138,12 @@ public class DataSinkNode extends OptimizerNode {
@Override
public void computeInterestingPropertiesForInputs(CostEstimator estimator) {
final InterestingProperties iProps = new InterestingProperties();
-
+
{
- final Ordering partitioning = getOperator().getPartitionOrdering();
- final DataDistribution dataDist = getOperator().getDataDistribution();
final RequestedGlobalProperties partitioningProps = new RequestedGlobalProperties();
- if (partitioning != null) {
- if(dataDist != null) {
- partitioningProps.setRangePartitioned(partitioning, dataDist);
- } else {
- partitioningProps.setRangePartitioned(partitioning);
- }
- iProps.addGlobalProperties(partitioningProps);
- }
iProps.addGlobalProperties(partitioningProps);
}
-
+
{
final Ordering localOrder = getOperator().getLocalOrder();
final RequestedLocalProperties orderProps = new RequestedLocalProperties();