[
https://issues.apache.org/jira/browse/FLINK-3234?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15138589#comment-15138589
]
ASF GitHub Bot commented on FLINK-3234:
---------------------------------------
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/1585#discussion_r52282174
--- Diff:
flink-java/src/main/java/org/apache/flink/api/java/operators/SortPartitionOperator.java
---
@@ -79,58 +119,41 @@ public SortPartitionOperator(DataSet<T> dataSet,
String sortField, Order sortOrd
* local partition sorting of the DataSet.
*
* @param field The field expression referring to the field of the
additional sort order of
- * the local partition sorting.
- * @param order The order of the additional sort order of the local
partition sorting.
+ * the local partition sorting.
+ * @param order The order of the additional sort order of the local
partition sorting.
* @return The DataSet with sorted local partitions.
*/
public SortPartitionOperator<T> sortPartition(String field, Order
order) {
- int[] flatOrderKeys = getFlatFields(field);
- this.appendSorting(flatOrderKeys, order);
+ if (useKeySelector) {
+ throw new InvalidProgramException("Expression keys
cannot be appended after a KeySelector");
+ }
+
+ ensureSortableKey(field);
+ keys.add(new Keys.ExpressionKeys<>(field, getType()));
+ orders.add(order);
+
return this;
}
- //
--------------------------------------------------------------------------------------------
- // Key Extraction
- //
--------------------------------------------------------------------------------------------
-
- private int[] getFlatFields(int field) {
+ public <K> SortPartitionOperator<T> sortPartition(KeySelector<T, K>
keyExtractor, Order order) {
+ throw new InvalidProgramException("KeySelector cannot be
chained.");
+ }
- if (!Keys.ExpressionKeys.isSortKey(field, super.getType())) {
+ private void ensureSortableKey(int field) throws
InvalidProgramException {
+ if (!Keys.ExpressionKeys.isSortKey(field, getType())) {
throw new InvalidProgramException("Selected sort key is
not a sortable type");
}
-
- Keys.ExpressionKeys<T> ek = new Keys.ExpressionKeys<>(field,
super.getType());
- return ek.computeLogicalKeyPositions();
}
- private int[] getFlatFields(String fields) {
-
- if (!Keys.ExpressionKeys.isSortKey(fields, super.getType())) {
+ private void ensureSortableKey(String field) throws
InvalidProgramException {
+ if (!Keys.ExpressionKeys.isSortKey(field, getType())) {
throw new InvalidProgramException("Selected sort key is
not a sortable type");
}
-
- Keys.ExpressionKeys<T> ek = new Keys.ExpressionKeys<>(fields,
super.getType());
- return ek.computeLogicalKeyPositions();
}
- private void appendSorting(int[] flatOrderFields, Order order) {
-
- if(this.sortKeyPositions == null) {
- // set sorting info
- this.sortKeyPositions = flatOrderFields;
- this.sortOrders = new Order[flatOrderFields.length];
- Arrays.fill(this.sortOrders, order);
- } else {
- // append sorting info to exising info
- int oldLength = this.sortKeyPositions.length;
- int newLength = oldLength + flatOrderFields.length;
- this.sortKeyPositions =
Arrays.copyOf(this.sortKeyPositions, newLength);
- this.sortOrders = Arrays.copyOf(this.sortOrders,
newLength);
-
- for(int i=0; i<flatOrderFields.length; i++) {
- this.sortKeyPositions[oldLength+i] =
flatOrderFields[i];
- this.sortOrders[oldLength+i] = order;
- }
+ private void ensureSortableKey(Keys<T> sortKey) {
--- End diff --
Change the `sortKey` parameter type to `SelectorFunctionKeys` and remove
the type check + cast.
> SortPartition does not support KeySelectorFunctions
> ---------------------------------------------------
>
> Key: FLINK-3234
> URL: https://issues.apache.org/jira/browse/FLINK-3234
> Project: Flink
> Issue Type: Improvement
> Components: DataSet API
> Affects Versions: 1.0.0, 0.10.1
> Reporter: Fabian Hueske
> Assignee: Chiwan Park
> Fix For: 1.0.0
>
>
> The following is not supported by the DataSet API:
> {code}
> DataSet<MyObject> data = ...
> DataSet<MyObject> data.sortPartition(
> new KeySelector<MyObject, Long>() {
> public Long getKey(MyObject v) {
> ...
> }
> },
> Order.ASCENDING);
> {code}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
