BatchStage.sort() #2469

MohamedMandouh · 2020-08-27T08:00:54Z

Introduces sorting feature for batch use cases of hazelcast jet.
Usage:
BatchStage.sort() for sorting based on natural ordering of items.
BatchStage.sort(ComparatorEx<V> comparator) for sorting based on user-defined comparator.

Breaking change: this changes the Edge and EdgeDef serialization format, breaking the backwards compatibility of a suspended job if it resumes on a newer Jet version.

Checklist

Tags Set
Milestone Set
New public APIs have @Nonnull/@Nullable annotations
New public APIs have @since tags in Javadoc

…y-sorting # Conflicts: # hazelcast-jet-core/src/main/java/com/hazelcast/jet/core/Edge.java # hazelcast-jet-core/src/main/java/com/hazelcast/jet/impl/execution/init/EdgeDef.java

devOpsHazelcast · 2020-08-27T08:00:56Z

Can one of the admins verify this patch?

devOpsHazelcast · 2020-08-27T08:00:56Z

Can one of the admins verify this patch?

devOpsHazelcast · 2020-08-27T08:00:56Z

Can one of the admins verify this patch?

devOpsHazelcast · 2020-08-27T08:00:56Z

Can one of the admins verify this patch?

devOpsHazelcast · 2020-08-27T08:00:56Z

Can one of the admins verify this patch?

devOpsHazelcast · 2020-08-27T08:00:56Z

Can one of the admins verify this patch?

@SInCE

add @SInCE cleanup

mtopolnik · 2020-08-31T09:51:46Z

...ast-jet-core/src/main/java/com/hazelcast/jet/impl/execution/ConcurrentInboundEdgeStream.java

+                    headItem = headObject;
+                }
+                if (headItem == null) {
+                    return NO_PROGRESS;


You may already have made progress by consuming some items in previous while loop iterations.

cangencer · 2020-08-31T18:14:12Z

This also needs adding to /site/docs/api/stateful-transforms.md

cangencer · 2020-08-31T18:15:06Z

hazelcast-jet-core/src/main/java/com/hazelcast/jet/core/Edge.java

+    }
+
+    /**
+     * Returns the comparator defined on this edge.


we could add a reference back to monoticOrder here

cangencer · 2020-08-31T18:16:29Z

hazelcast-jet-core/src/main/java/com/hazelcast/jet/pipeline/BatchStage.java

+     * @since 4.3
+     */
+    @Nonnull
+    BatchStage<T> sort(@Nonnull ComparatorEx<T> comparator);


should be ? super T

cangencer · 2020-08-31T18:16:56Z

hazelcast-jet-core/src/main/java/com/hazelcast/jet/impl/pipeline/transform/SortTransform.java

+import static com.hazelcast.jet.core.processor.Processors.sortPrepareP;
+
+
+public class SortTransform<V> extends AbstractTransform {


should use T here instead of V

cangencer · 2020-08-31T18:19:01Z

hazelcast-jet-core/src/main/java/com/hazelcast/jet/impl/processor/SortPrepareP.java

+import java.util.TreeSet;
+
+public class SortPrepareP<V> extends AbstractProcessor {
+    private final SortedSet<V> set;


I don't think using SortedSet is correct here. It will skip duplicates. PriorityQueue would be more correct.

hazelcast-jet-core/src/main/java/com/hazelcast/jet/impl/processor/SortP.java

hazelcast-jet-core/src/main/java/com/hazelcast/jet/core/processor/Processors.java

hazelcast-jet-core/src/main/java/com/hazelcast/jet/pipeline/BatchStage.java

cangencer · 2020-09-14T05:18:54Z

@MohamedMandouh if possible, you should do partialSort in another PR to keep the scope small for this one.

mtopolnik · 2020-09-14T11:41:26Z

Just stage.aggregate(topN(10)) will give you a sorted list as output, there is no need for any other upstream action. Theoretically, for optimization purposes, we could have every upstream processor limit its output to 10 items, and this could then become a higher-level stage operation that could be captured with apply.

This reverts commit c0d102b

This reverts commit 0e38170

mtopolnik · 2020-09-16T12:01:45Z

verify

devOpsHazelcast · 2020-09-16T12:20:15Z

Can one of the admins verify this patch?

devOpsHazelcast · 2020-09-16T12:20:15Z

Can one of the admins verify this patch?

devOpsHazelcast · 2020-09-16T12:20:15Z

Can one of the admins verify this patch?

devOpsHazelcast · 2020-09-16T12:20:15Z

Can one of the admins verify this patch?

devOpsHazelcast · 2020-09-16T12:20:16Z

Can one of the admins verify this patch?

devOpsHazelcast · 2020-09-16T12:20:16Z

Can one of the admins verify this patch?

MohamedMandouh added 6 commits August 26, 2020 15:39

add in-memory sort

cfcd5fe

add in-memory sort

25cfa0f

Merge remote-tracking branch 'origin/in-memory-sorting' into in-memor…

7478050

…y-sorting # Conflicts: # hazelcast-jet-core/src/main/java/com/hazelcast/jet/core/Edge.java # hazelcast-jet-core/src/main/java/com/hazelcast/jet/impl/execution/init/EdgeDef.java

remove SerializableComparator

26e76f7

replace TreeMap with TreeSet

139910d

replace ComparatorEx by Comparator in SortPrepareP

b7e2393

mtopolnik added enhancement Pipeline API labels Aug 27, 2020

mtopolnik added this to the 4.3 milestone Aug 27, 2020

mtopolnik requested a review from viliam-durina August 27, 2020 08:38

MohamedMandouh added 2 commits August 27, 2020 11:21

add @NonNull/@nullable

8d4013f

add @SInCE cleanup

add BatchStage.sort()

a1e488d

mtopolnik reviewed Aug 31, 2020

View reviewed changes

MohamedMandouh added 2 commits August 31, 2020 17:56

fix returning NO_PROGRESS whenever a queue returns null

1176d43

pass checkstyle

4c8897d

cangencer reviewed Aug 31, 2020

View reviewed changes

MohamedMandouh mentioned this pull request Sep 1, 2020

RocksDB State Backend #2257

Closed

apply requested changes

9f4a0b1

MohamedMandouh requested a review from mtopolnik September 2, 2020 11:30

Add @nonnull

9b9cba5

Marko Topolnik and others added 3 commits September 5, 2020 15:53

Fix two bugs

8cb3897

Add BatchStageTest.sort()

f1bbf30

add ProcessorsTest.sort

d629a5c

ufukyilmaz reviewed Sep 7, 2020

View reviewed changes

hazelcast-jet-core/src/main/java/com/hazelcast/jet/impl/processor/SortP.java Show resolved Hide resolved

ufukyilmaz reviewed Sep 7, 2020

View reviewed changes

hazelcast-jet-core/src/main/java/com/hazelcast/jet/core/processor/Processors.java Show resolved Hide resolved

ufukyilmaz reviewed Sep 7, 2020

View reviewed changes

hazelcast-jet-core/src/main/java/com/hazelcast/jet/pipeline/BatchStage.java Show resolved Hide resolved

MohamedMandouh added 5 commits September 8, 2020 17:27

add missing Override

a23a817

add sample usage, param and return annotaitions

eb6751d

add "only available for batch" in javadoc

5f59a0d

introduce BatchStage.partialSort

0e38170

cleanup

c0d102b

MohamedMandouh and others added 4 commits September 14, 2020 15:25

Revert "cleanup"

029432b

This reverts commit c0d102b

Revert "introduce BatchStage.partialSort"

a045a23

This reverts commit 0e38170

Improve Javadoc

b3d124d

Improved Edge doc

85a31da

mtopolnik approved these changes Sep 14, 2020

View reviewed changes

Add jet-start.sh docs

befdc1d

mtopolnik approved these changes Sep 16, 2020

View reviewed changes

mtopolnik merged commit 73fe09d into hazelcast:master Sep 16, 2020

mtopolnik changed the title ~~In-Memory Sorting~~ BatchStage.sort() Sep 16, 2020

mtopolnik added the breaking-change-core label Oct 2, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

BatchStage.sort() #2469

BatchStage.sort() #2469

MohamedMandouh commented Aug 27, 2020 •

edited by mtopolnik

devOpsHazelcast commented Aug 27, 2020

devOpsHazelcast commented Aug 27, 2020

devOpsHazelcast commented Aug 27, 2020

devOpsHazelcast commented Aug 27, 2020

devOpsHazelcast commented Aug 27, 2020

devOpsHazelcast commented Aug 27, 2020

mtopolnik Aug 31, 2020

MohamedMandouh Aug 31, 2020

cangencer commented Aug 31, 2020

cangencer Aug 31, 2020

cangencer Aug 31, 2020

cangencer Aug 31, 2020

cangencer Aug 31, 2020

cangencer commented Sep 14, 2020

mtopolnik commented Sep 14, 2020

mtopolnik commented Sep 16, 2020

devOpsHazelcast commented Sep 16, 2020

devOpsHazelcast commented Sep 16, 2020

devOpsHazelcast commented Sep 16, 2020

devOpsHazelcast commented Sep 16, 2020

devOpsHazelcast commented Sep 16, 2020

devOpsHazelcast commented Sep 16, 2020

		import static com.hazelcast.jet.core.processor.Processors.sortPrepareP;


		public class SortTransform<V> extends AbstractTransform {

BatchStage.sort() #2469

BatchStage.sort() #2469

Conversation

MohamedMandouh commented Aug 27, 2020 • edited by mtopolnik

devOpsHazelcast commented Aug 27, 2020

devOpsHazelcast commented Aug 27, 2020

devOpsHazelcast commented Aug 27, 2020

devOpsHazelcast commented Aug 27, 2020

devOpsHazelcast commented Aug 27, 2020

devOpsHazelcast commented Aug 27, 2020

mtopolnik Aug 31, 2020

Choose a reason for hiding this comment

MohamedMandouh Aug 31, 2020

Choose a reason for hiding this comment

cangencer commented Aug 31, 2020

cangencer Aug 31, 2020

Choose a reason for hiding this comment

cangencer Aug 31, 2020

Choose a reason for hiding this comment

cangencer Aug 31, 2020

Choose a reason for hiding this comment

cangencer Aug 31, 2020

Choose a reason for hiding this comment

cangencer commented Sep 14, 2020

mtopolnik commented Sep 14, 2020

mtopolnik commented Sep 16, 2020

devOpsHazelcast commented Sep 16, 2020

devOpsHazelcast commented Sep 16, 2020

devOpsHazelcast commented Sep 16, 2020

devOpsHazelcast commented Sep 16, 2020

devOpsHazelcast commented Sep 16, 2020

devOpsHazelcast commented Sep 16, 2020

MohamedMandouh commented Aug 27, 2020 •

edited by mtopolnik