#52 CQuery: fix distinct & flat

assira.core/src/main/java/net/ranides/assira/collection/query/CQueryAbstract.java

@@ -10,7 +10,6 @@ import net.ranides.assira.collection.HashFunction;
 import net.ranides.assira.collection.iterators.IterableUtils;
 import net.ranides.assira.collection.query.base.CQList;
 import net.ranides.assira.collection.query.derived.*;
-import net.ranides.assira.collection.query.support.BaseArray;
 import net.ranides.assira.collection.query.support.BaseEach;
 import net.ranides.assira.collection.query.support.BaseFeatures;
 import net.ranides.assira.collection.query.support.BaseIterable;
@@ -328,17 +327,17 @@ public abstract class CQueryAbstract<T> implements CQuery<T>, CQueryFeatures {
 
     @Override
     public <R> CQuery<R> flat(SplitQueryFunction<? super T, R> f) {
-        return BaseSplit.split(this, f);
+        return BaseSplit.flat(this, f);
     }
 
     @Override
     public <R> CQuery<R> flatIterable(SplitIterableFunction<? super T, R> f) {
-        return BaseSplit.split(this, f);
+        return BaseSplit.flat(this, f);
     }
 
     @Override
     public <R> CQuery<R> flatArray(SplitArrayFunction<? super T, R> f) {
-        return BaseSplit.split(this, f);
+        return BaseSplit.flat(this, f);
     }
 
     @Override
@@ -457,17 +456,17 @@ public abstract class CQueryAbstract<T> implements CQuery<T>, CQueryFeatures {
 
     @Override
     public CQuery<T> distinct() {
-        return filter(new OpenSet<T>()::add);
+        return new CQFilter<>(this, () -> new OpenSet<>()::add);
     }
 
     @Override
     public CQuery<T> distinct(HashFunction<T> eq) {
-        return filter(new CustomSet<>(eq)::add);
+        return new CQFilter<>(this, () -> new CustomSet<>(eq)::add);
     }
 
     @Override
     public CQuery<T> distinct(Comparator<? super T> cmp) {
-        return filter(new TreeSet<>(cmp)::add);
+        return new CQFilter<>(this, () -> new TreeSet<>(cmp)::add);
     }
 
     @Override

assira.core/src/main/java/net/ranides/assira/collection/query/derived/CQFilter.java

@@ -7,33 +7,41 @@ import net.ranides.assira.functional.Predicates;
 
 import java.util.Iterator;
 import java.util.function.Predicate;
+import java.util.function.Supplier;
 import java.util.stream.Stream;
 
 public class CQFilter<T> extends CQAbstractFilter<T,T> {
-    private final Predicate<? super T> p;
+    private final Supplier<Predicate<? super T>> p;
 
-    public CQFilter(CQueryAbstract<T> source, Predicate<? super T> p) {
+    public CQFilter(CQueryAbstract<T> source, Predicate<? super T> predicate) {
         super(source);
-        this.p = p;
+        this.p = () -> predicate;
+    }
+
+    public CQFilter(CQueryAbstract<T> source, Supplier<Predicate<? super T>> predicate) {
+        super(source);
+        this.p = predicate;
     }
 
     @Override
     public Stream<T> stream() {
-        return source.stream().filter(p);
+        return source.stream().filter(p.get());
     }
 
     @Override
     public Iterator<T> iterator() {
-        return IteratorUtils.filter(source.iterator(), p);
+        return IteratorUtils.filter(source.iterator(), p.get());
     }
 
     @Override
     public boolean whileEach(Predicate<? super T> consumer) {
-        return source.whileEach(v -> !p.test(v) || consumer.test(v));
+        Predicate<? super T> predicate = p.get();
+        return source.whileEach(v -> !predicate.test(v) || consumer.test(v));
     }
 
     @Override
     public boolean whileEach(Predicates.EachPredicate<? super T> consumer) {
-        return source.whileEach((i,v) -> !p.test(v) || consumer.test(i,v));
+        Predicate<? super T> predicate = p.get();
+        return source.whileEach((i,v) -> !predicate.test(v) || consumer.test(i,v));
     }
 }

assira.core/src/main/java/net/ranides/assira/collection/query/support/BaseSplit.java

@@ -7,15 +7,43 @@ import net.ranides.assira.collection.query.CQueryAbstract;
 import net.ranides.assira.collection.query.base.CQAbstractFilter;
 
 import java.util.Iterator;
+import java.util.List;
 import java.util.function.Predicate;
 import java.util.stream.Stream;
 
 public class BaseSplit {
 
-    public static <T, R> CQuery<R> split(CQueryAbstract<T> that, CQuery.SplitQueryFunction<? super T, R> f) {
-        return new CQAbstractFilter<T,R>(that) {
+    // Implementation note:
+    //
+    // We can't use stream#flatMap because derived stream evaluates too much.
+    // more precisely, it applies #map to all elements generated from element:
+    // EXPECTED:
+    //      A source:   abc def ghi
+    //      B flat:     a b c | d e f | .
+    //      C map:      A B C | D . .
+    //      d limit(4)  A B C | D
+    // UNFORTUNATELY:
+    //      A source:   abc def ghi
+    //      B flat:     a b c | d e f | .
+    //      C map:      A B C | D E F
+    //      d limit(4)  A B C | D
+    //
+    // So we fall-back to our own iterator implementation, sorry
+    //
+    // Reference:
+    //  https://bugs.openjdk.java.net/browse/JDK-8075939
+    //  https://stackoverflow.com/questions/29229373/why-filter-after-flatmap-is-not-completely-lazy-in-java-streams
+
+
+    private static final boolean DONT_USE_FLAT_STREAM = true;
+
+    public static <T, R> CQuery<R> flat(CQueryAbstract<T> that, CQuery.SplitQueryFunction<? super T, R> f) {
+        return new CQAbstractSplit<T,R>(that) {
             @Override
             public Stream<R> stream() {
+                if (DONT_USE_FLAT_STREAM) {
+                    return BaseIterable.stream(this);
+                }
                 return source.stream().flatMap(s -> f.apply(s).stream());
             }
 
@@ -28,13 +56,21 @@ public class BaseSplit {
             public boolean whileEach(Predicate<? super R> consumer) {
                 return source.whileEach(s -> f.apply(s).whileEach(consumer));
             }
+
+            @Override
+            public List<R> list() {
+                return BaseStream.list(this);
+            }
         };
     }
 
-    public static <T,R> CQuery<R> split(CQueryAbstract<T> that, CQuery.SplitIterableFunction<? super T, R> f) {
-        return new CQAbstractFilter<T,R>(that) {
+    public static <T,R> CQuery<R> flat(CQueryAbstract<T> that, CQuery.SplitIterableFunction<? super T, R> f) {
+        return new CQAbstractSplit<T,R>(that) {
             @Override
             public Stream<R> stream() {
+                if (DONT_USE_FLAT_STREAM) {
+                    return BaseIterable.stream(this);
+                }
                 return source.stream().flatMap(s -> IteratorUtils.stream(f.apply(s).iterator()));
             }
 
@@ -51,10 +87,13 @@ public class BaseSplit {
         };
     }
 
-    public static <T,R> CQuery<R> split(CQueryAbstract<T> that, CQuery.SplitArrayFunction<? super T, R> f) {
-        return new CQAbstractFilter<T,R>(that) {
+    public static <T,R> CQuery<R> flat(CQueryAbstract<T> that, CQuery.SplitArrayFunction<? super T, R> f) {
+        return new CQAbstractSplit<T,R>(that) {
             @Override
             public Stream<R> stream() {
+                if (DONT_USE_FLAT_STREAM) {
+                    return BaseIterable.stream(this);
+                }
                 return source.stream().flatMap(s -> Stream.of(f.apply(s)));
             }
 
@@ -72,9 +111,12 @@ public class BaseSplit {
     }
 
     public static <T> CQuery<T> splitIf(CQueryAbstract<T> that, Predicate<? super T> p, CQuery.SplitQueryFunction<? super T, T> f) {
-        return new CQAbstractFilter<T,T>(that) {
+        return new CQAbstractSplit<T,T>(that) {
             @Override
             public Stream<T> stream() {
+                if (DONT_USE_FLAT_STREAM) {
+                    return BaseIterable.stream(this);
+                }
                 return source.stream().flatMap(s -> p.test(s) ? f.apply(s).stream() : Stream.of(s));
             }
 
@@ -95,9 +137,12 @@ public class BaseSplit {
     }
 
     public static <T> CQuery<T> splitIf(CQueryAbstract<T> that, Predicate<? super T> p, CQuery.SplitIterableFunction<? super T, T> f) {
-        return new CQAbstractFilter<T,T>(that) {
+        return new CQAbstractSplit<T,T>(that) {
             @Override
             public Stream<T> stream() {
+                if (DONT_USE_FLAT_STREAM) {
+                    return BaseIterable.stream(this);
+                }
                 return source.stream().flatMap(s -> {
                     return p.test(s) ? IteratorUtils.stream(f.apply(s).iterator()) : Stream.of(s);
                 });
@@ -121,9 +166,12 @@ public class BaseSplit {
     }
 
     public static <T> CQuery<T> splitIf(CQueryAbstract<T> that, Predicate<? super T> p, CQuery.SplitArrayFunction<? super T, T> f) {
-        return new CQAbstractFilter<T,T>(that) {
+        return new CQAbstractSplit<T,T>(that) {
             @Override
             public Stream<T> stream() {
+                if (DONT_USE_FLAT_STREAM) {
+                    return BaseIterable.stream(this);
+                }
                 return source.stream().flatMap(s -> {
                     return p.test(s) ? Stream.of(f.apply(s)) : Stream.of(s);
                 });
@@ -146,4 +194,16 @@ public class BaseSplit {
         };
     }
 
+    private static abstract class CQAbstractSplit<T,R> extends CQAbstractFilter<T,R> {
+
+        public CQAbstractSplit(CQueryAbstract<T> source) {
+            super(source);
+        }
+
+        @Override
+        public boolean hasFastStream() {
+            return !DONT_USE_FLAT_STREAM && source.hasFastStream();
+        }
+    }
+
 }

assira.core/src/test/java/net/ranides/assira/reflection/util/ServiceScannerTest.java

@@ -37,7 +37,6 @@ public class ServiceScannerTest {
 
     @Test
     public void lazy() {
-        // to jest ważny błąd w semantyce CQuery! Coś leniwie obliczanego nie jest leniwe
         List<String> names = new ArrayList<>();
 
         CQuery<URIHandler> query = ServiceScanner.scanner(URIHandler.class)