github/jooq
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[jOOQ/jOOQ#11895] ResultQuery.fetchStream() fetches entire resultset at once

Browse Source
This commit is contained in:
Lukas Eder 2021-05-17 17:54:43 +02:00
parent bb3f3e30ea
commit ed5ed02be2
2 changed files with 21 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
jOOQ/src/main/java/org/jooq/impl/FieldsImpl.java
View File

@ -97,7 +97,7 @@ final class FieldsImpl<R extends Record> extends AbstractQueryPart implements Re
@Override
public final <U> RecordMapper<R, U> mapper(int fieldIndex, Configuration configuration, Class<? extends U> type) {
return mapper(fieldIndex, configuration.converterProvider().provide(field(fieldIndex).getType(), type));
return mapper(fieldIndex, converterOrFail(configuration, field(fieldIndex).getType(), type));
}
@Override

21
jOOQ/src/main/java/org/jooq/impl/ResultQueryTrait.java
View File

@ -42,6 +42,7 @@ import static org.jooq.Records.intoList;
import static org.jooq.Records.intoMap;
import static org.jooq.Records.intoSet;
import static org.jooq.impl.Tools.blocking;
import static org.jooq.tools.jdbc.JDBCUtils.safeClose;
import java.lang.reflect.Array;
import java.sql.ResultSet;
@ -53,11 +54,14 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.Spliterator;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collector;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;
import org.jooq.Configuration;
import org.jooq.Converter;
@ -99,6 +103,7 @@ import org.jooq.exception.DataAccessException;
import org.jooq.impl.R2DBC.BlockingRecordSubscription;
import org.jooq.impl.R2DBC.QuerySubscription;
import org.jooq.impl.R2DBC.ResultSubscriber;
import org.jooq.tools.jdbc.JDBCUtils;
import org.reactivestreams.Subscriber;
@ -293,7 +298,21 @@ interface ResultQueryTrait<R extends Record> extends QueryPartInternal, ResultQu
@Override
default Stream<R> fetchStream() {
return Stream.of(1).flatMap(i -> fetchLazy().stream());
AtomicReference<Cursor<R>> r = new AtomicReference<>();
// [#11895] Don't use the Stream.of(1).flatMap(i -> fetchLazy().stream())
// trick, because flatMap() will consume the entire result set
return StreamSupport.stream(
() -> {
Cursor<R> c = fetchLazy();
r.set(c);
return c.spliterator();
},
Spliterator.IMMUTABLE | Spliterator.NONNULL | Spliterator.ORDERED,
false
).onClose(() -> {
safeClose(r.get());
});
}
@Override