Skip to content

Support for Streams and Iterators #2067

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
bbakerman merged 1 commit into from
Oct 9, 2020
Merged

Support for Streams and Iterators #2067

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 13 additions & 10 deletions src/main/java/graphql/execution/ExecutionStrategy.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,10 +39,10 @@
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.OptionalInt;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.function.Supplier;
Expand All @@ -56,7 +56,9 @@
import static graphql.execution.FieldValueInfo.CompleteValueType.OBJECT;
import static graphql.execution.FieldValueInfo.CompleteValueType.SCALAR;
import static graphql.schema.DataFetchingEnvironmentImpl.newDataFetchingEnvironment;
import static graphql.schema.GraphQLTypeUtil.isEnum;
import static graphql.schema.GraphQLTypeUtil.isList;
import static graphql.schema.GraphQLTypeUtil.isScalar;
import static java.util.concurrent.CompletableFuture.completedFuture;

/**
Expand Down Expand Up @@ -421,10 +423,10 @@ protected FieldValueInfo completeValue(ExecutionContext executionContext, Execut
return FieldValueInfo.newFieldValueInfo(NULL).fieldValue(fieldValue).build();
} else if (isList(fieldType)) {
return completeValueForList(executionContext, parameters, result);
} else if (fieldType instanceof GraphQLScalarType) {
} else if (isScalar(fieldType)) {
fieldValue = completeValueForScalar(executionContext, parameters, (GraphQLScalarType) fieldType, result);
return FieldValueInfo.newFieldValueInfo(SCALAR).fieldValue(fieldValue).build();
} else if (fieldType instanceof GraphQLEnumType) {
} else if (isEnum(fieldType)) {
fieldValue = completeValueForEnum(executionContext, parameters, (GraphQLEnumType) fieldType, result);
return FieldValueInfo.newFieldValueInfo(ENUM).fieldValue(fieldValue).build();
}
Expand Down Expand Up @@ -494,19 +496,19 @@ protected FieldValueInfo completeValueForList(ExecutionContext executionContext,
*/
protected FieldValueInfo completeValueForList(ExecutionContext executionContext, ExecutionStrategyParameters parameters, Iterable<Object> iterableValues) {

Collection<Object> values = FpKit.toCollection(iterableValues);
OptionalInt size = FpKit.toSize(iterableValues);
ExecutionStepInfo executionStepInfo = parameters.getExecutionStepInfo();

InstrumentationFieldCompleteParameters instrumentationParams = new InstrumentationFieldCompleteParameters(executionContext, parameters, () -> executionStepInfo, values);
InstrumentationFieldCompleteParameters instrumentationParams = new InstrumentationFieldCompleteParameters(executionContext, parameters, () -> executionStepInfo, iterableValues);

@bbakerman bbakerman Oct 6, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is a mild breaking change - the input is "Object" but the actual type has changed.

I think this is ok (did we say it WOULD always be a collection??) but I just want to call it out and see others thoughts on this.

@dfa1 dfa1 Oct 6, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes, this could be a breaking change.
Arrays are reallocated anymore as List (but I could easily re-introduce this behavior).

For Collections we should be fine: toIterable does not change the concrete type.
For Streams/Iterators we should be fine as well: nobody is using that right now.

Instrumentation instrumentation = executionContext.getInstrumentation();

InstrumentationContext<ExecutionResult> completeListCtx = instrumentation.beginFieldListComplete(
instrumentationParams
);

List<FieldValueInfo> fieldValueInfos = new ArrayList<>(values.size());
List<FieldValueInfo> fieldValueInfos = new ArrayList<>(size.orElse(1));

@bbakerman bbakerman Oct 6, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

At the end of the day we are still allocated an array the size of the list - So source of the data can be a stream but we can stay streaming all the way.

This is an improvement - as you say it greatly reduces your memory usage (nearly half??) but longer term could we have streams all the way down to the turtles?

Thinking aloud here it might be too theoretical - I know there is code out there that assumes ExecutionResult is really a Map of lists and maps and values. Would introducing "streams" in that break things?

@dfa1 dfa1 Oct 6, 2020 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

At the end of the day we are still allocated an array the size of the list - So source of the data can be a stream but we can stay streaming all the way.

yes, I was not able to remove this big list.

This is an improvement - as you say it greatly reduces your memory usage (nearly half??) but longer term could we have streams all the way down to the turtles?
Don't know yet: eventually everything is "materialized" in a single ExecutionResult. Would be nice to minimize everything else.

This the most big query we have so far.

Before
image

After
image

Thinking aloud here it might be too theoretical - I know there is code out there that assumes ExecutionResult is really a Map of lists and maps and values. Would introducing "streams" in that break things?

We are merely introducing streaming between datafetchers and execution strategy: so this is very conservative choice by now.

int index = 0;
for (Object item : values) {
for (Object item : iterableValues) {
ResultPath indexedPath = parameters.getPath().segment(index);

ExecutionStepInfo stepInfoForListElement = executionStepInfoFactory.newExecutionStepInfoForListElement(executionStepInfo, index);
Expand All @@ -519,7 +521,7 @@ protected FieldValueInfo completeValueForList(ExecutionContext executionContext,
ExecutionStrategyParameters newParameters = parameters.transform(builder ->
builder.executionStepInfo(stepInfoForListElement)
.nonNullFieldValidator(nonNullableFieldValidator)
.listSize(values.size())
.listSize(size.orElse(-1)) // -1 signals that we don't know the size

@bbakerman bbakerman Oct 6, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I had a look at graphql.execution.ExecutionStrategyParameters#getListSize - we never use it.

I was worried this -1 semantics could affect some one )badly but we never use it - I think this will be ok

@dfa1 dfa1 Oct 6, 2020 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes, I think it is backward compatible.

Maybe we can introduce a new method that returns OptionalInt ?

.localContext(value.getLocalContext())
.currentListIndex(finalIndex)
.path(indexedPath)
Expand Down Expand Up @@ -674,14 +676,15 @@ protected GraphQLObjectType resolveType(ExecutionContext executionContext, Execu


protected Iterable<Object> toIterable(ExecutionContext context, ExecutionStrategyParameters parameters, Object result) {
if (result.getClass().isArray() || result instanceof Iterable) {
return toIterable(result);
if (FpKit.isIterable(result)) {
return FpKit.toIterable(result);
}

handleTypeMismatchProblem(context, parameters, result);
return null;
}


private void handleTypeMismatchProblem(ExecutionContext context, ExecutionStrategyParameters parameters, Object result) {
TypeMismatchError error = new TypeMismatchError(parameters.getPath(), parameters.getExecutionStepInfo().getUnwrappedNonNullType());
logNotSafe.warn("{} got {}", error.getMessage(), result.getClass());
Expand Down
107 changes: 86 additions & 21 deletions src/main/java/graphql/util/FpKit.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,9 @@
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Optional;
import java.util.OptionalInt;
import java.util.concurrent.CompletableFuture;
import java.util.function.BiFunction;
import java.util.function.BinaryOperator;
Expand All @@ -19,6 +21,7 @@
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;

import static java.util.Collections.singletonList;
import static java.util.function.Function.identity;
Expand All @@ -31,10 +34,10 @@ public class FpKit {
// From a list of named things, get a map of them by name, merging them according to the merge function
public static <T> Map<String, T> getByName(List<T> namedObjects, Function<T, String> nameFn, BinaryOperator<T> mergeFunc) {
return namedObjects.stream().collect(Collectors.toMap(
nameFn,
identity(),
mergeFunc,
LinkedHashMap::new)
nameFn,
identity(),
mergeFunc,
LinkedHashMap::new)
);
}

Expand All @@ -45,10 +48,10 @@ public static <T, NewKey> Map<NewKey, List<T>> groupingBy(Collection<T> list, Fu

public static <T, NewKey> Map<NewKey, T> groupingByUniqueKey(Collection<T> list, Function<T, NewKey> keyFunction) {
return list.stream().collect(Collectors.toMap(
keyFunction,
identity(),
throwingMerger(),
LinkedHashMap::new)
keyFunction,
identity(),
throwingMerger(),
LinkedHashMap::new)
);
}

Expand All @@ -75,17 +78,15 @@ public static <T> BinaryOperator<T> mergeFirst() {
*
* @param iterableResult the result object
* @param <T> the type of thing
*
* @return an Iterable from that object
*
* @throws java.lang.ClassCastException if its not an Iterable
*/
@SuppressWarnings("unchecked")
public static <T> Collection<T> toCollection(Object iterableResult) {
if (iterableResult.getClass().isArray()) {
List<Object> collect = IntStream.range(0, Array.getLength(iterableResult))
.mapToObj(i -> Array.get(iterableResult, i))
.collect(Collectors.toList());
.mapToObj(i -> Array.get(iterableResult, i))
.collect(Collectors.toList());
return (List<T>) collect;
}
if (iterableResult instanceof Collection) {
Expand All @@ -100,14 +101,79 @@ public static <T> Collection<T> toCollection(Object iterableResult) {
return list;
}

public static boolean isIterable(Object result) {
return result.getClass().isArray() || result instanceof Iterable || result instanceof Stream || result instanceof Iterator;
}


@SuppressWarnings("unchecked")
public static <T> Iterable<T> toIterable(Object iterableResult) {
if (iterableResult instanceof Iterable) {
return ((Iterable<T>) iterableResult);
}

if (iterableResult instanceof Stream) {
return ((Stream<T>) iterableResult)::iterator;
}

if (iterableResult instanceof Iterator) {
return () -> (Iterator<T>) iterableResult;
}

if (iterableResult.getClass().isArray()) {
return () -> new ArrayIterator<>(iterableResult);
}

throw new ClassCastException("not Iterable: " + iterableResult.getClass());
}

private static class ArrayIterator<T> implements Iterator<T> {

private final Object array;
private final int size;
private int i;

private ArrayIterator(Object array) {
this.array = array;
this.size = Array.getLength(array);
this.i = 0;
}

@Override
public boolean hasNext() {
return i < size;
}

@SuppressWarnings("unchecked")
@Override
public T next() {
if (!hasNext()) {
throw new NoSuchElementException();
}
return (T) Array.get(array, i++);
}

}

public static OptionalInt toSize(Object iterableResult) {
if (iterableResult instanceof Collection) {
return OptionalInt.of(((Collection<?>) iterableResult).size());
}

if (iterableResult.getClass().isArray()) {
return OptionalInt.of(Array.getLength(iterableResult));
}

return OptionalInt.empty();
}

/**
* Concatenates (appends) a single elements to an existing list
*
* @param l the list onto which to append the element
* @param t the element to append
* @param <T> the type of elements of the list
*
* @return a <strong>new</strong> list componsed of the first list elements and the new element
* @return a <strong>new</strong> list composed of the first list elements and the new element
*/
public static <T> List<T> concat(List<T> l, T t) {
return concat(l, singletonList(t));
Expand All @@ -119,7 +185,6 @@ public static <T> List<T> concat(List<T> l, T t) {
* @param l1 the first list to concatenate
* @param l2 the second list to concatenate
* @param <T> the type of element of the lists
*
* @return a <strong>new</strong> list composed of the two concatenated lists elements
*/
public static <T> List<T> concat(List<T> l1, List<T> l2) {
Expand Down Expand Up @@ -152,7 +217,7 @@ public static <T> List<List<T>> transposeMatrix(List<? extends List<T>> matrix)
for (int j = 0; j < colCount; j++) {
T val = matrix.get(i).get(j);
if (result.size() <= j) {
result.add(j, new ArrayList());
result.add(j, new ArrayList<>());
}
result.get(j).add(i, val);
}
Expand All @@ -166,15 +231,15 @@ public static <T> CompletableFuture<List<T>> flatList(CompletableFuture<List<Lis

public static <T> List<T> flatList(List<List<T>> listLists) {
return listLists.stream()
.flatMap(List::stream)
.collect(Collectors.toList());
.flatMap(List::stream)
.collect(Collectors.toList());
}

public static <T> Optional<T> findOne(Collection<T> list, Predicate<T> filter) {
return list
.stream()
.filter(filter)
.findFirst();
.stream()
.filter(filter)
.findFirst();
}

public static <T> T findOneOrNull(List<T> list, Predicate<T> filter) {
Expand Down
50 changes: 50 additions & 0 deletions src/test/groovy/graphql/execution/ExecutionStrategyTest.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ import spock.lang.Specification

import java.util.concurrent.CompletableFuture
import java.util.concurrent.CompletionException
import java.util.stream.Stream

import static ExecutionStrategyParameters.newParameters
import static graphql.Scalars.GraphQLString
Expand Down Expand Up @@ -690,6 +691,55 @@ class ExecutionStrategyTest extends Specification {
executionResult.get().data == [1L, 2L, 3L]
}

def "#842 completes value for java.util.Stream"() {
given:
ExecutionContext executionContext = buildContext()
Stream<Long> result = Stream.of(1L, 2L, 3L)
def fieldType = list(Scalars.GraphQLLong)
def fldDef = newFieldDefinition().name("test").type(fieldType).build()
def executionStepInfo = ExecutionStepInfo.newExecutionStepInfo().type(fieldType).path(ResultPath.rootPath()).fieldDefinition(fldDef).build()
NonNullableFieldValidator nullableFieldValidator = new NonNullableFieldValidator(executionContext, executionStepInfo)

def parameters = newParameters()
.executionStepInfo(executionStepInfo)
.source(result)
.nonNullFieldValidator(nullableFieldValidator)
.fields(mergedSelectionSet(["fld": [mergedField(Field.newField().build())]]))
.field(mergedField(Field.newField().build()))
.build()

when:
def executionResult = executionStrategy.completeValue(executionContext, parameters).fieldValue

then:
executionResult.get().data == [1L, 2L, 3L]
}

def "#842 completes value for java.util.Iterator"() {
given:
ExecutionContext executionContext = buildContext()
Iterator<Long> result = Arrays.asList(1L, 2L, 3L).iterator()
def fieldType = list(Scalars.GraphQLLong)
def fldDef = newFieldDefinition().name("test").type(fieldType).build()
def executionStepInfo = ExecutionStepInfo.newExecutionStepInfo().type(fieldType).path(ResultPath.rootPath()).fieldDefinition(fldDef).build()
NonNullableFieldValidator nullableFieldValidator = new NonNullableFieldValidator(executionContext, executionStepInfo)

def parameters = newParameters()
.executionStepInfo(executionStepInfo)
.source(result)
.nonNullFieldValidator(nullableFieldValidator)
.fields(mergedSelectionSet(["fld": [mergedField(Field.newField().build())]]))
.field(mergedField(Field.newField().build()))
.build()

when:
def executionResult = executionStrategy.completeValue(executionContext, parameters).fieldValue

then:
executionResult.get().data == [1L, 2L, 3L]
}


def "#820 processes DataFetcherResult"() {
given:

Expand Down