rubenada commented on a change in pull request #2080:
URL: https://github.com/apache/calcite/pull/2080#discussion_r460172663
##########
File path:
linq4j/src/main/java/org/apache/calcite/linq4j/EnumerableDefaults.java
##########
@@ -2636,6 +2636,56 @@ public static boolean isMergeJoinSupported(JoinType
joinType) {
return orderBy(source, keySelector, Collections.reverseOrder(comparator));
}
+ /**
+ * A sort implementation based on a heap (or priority queue).
+ * @param fetch must be greater than 0
+ * @param offset must be greater than or equal to 0
+ */
+ public static <TSource, TKey> Enumerable<TSource> orderBy(
+ Enumerable<TSource> source,
+ Function1<TSource, TKey> keySelector,
+ Comparator<TKey> comparator,
+ int fetch,
+ int offset) {
+ return new AbstractEnumerable<TSource>() {
+ @Override public Enumerator<TSource> enumerator() {
+ TopNHeap<TSource, TKey> heap = new TopNHeap<>(
+ keySelector,
+ comparator,
+ fetch,
+ offset);
+ try (Enumerator<TSource> os = source.enumerator()) {
+ while (os.moveNext()) {
+ TSource o = os.current();
+ heap.offer(o);
+ }
+ }
+
+ // reverse queue
+ Object[] result = heap.getResult();
+ return new Enumerator<TSource>() {
+ int i = -1;
+
+ @SuppressWarnings("unchecked")
+ @Override public TSource current() {
+ return (TSource) result[this.i];
+ }
+
+ @Override public boolean moveNext() {
+ return ++this.i < result.length;
+ }
+
+ @Override public void reset() {
+ this.i = -1;
+ }
+
+ @Override public void close() {
+ }
Review comment:
Should we close here `source`?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
