Skip to content
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

Introduce Iterators#flatMap #92202

Merged
merged 3 commits into from
Dec 7, 2022
Merged

Introduce Iterators#flatMap #92202

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
73ad403
Introduce Iterators#flatMap
DaveCTurner Dec 7, 2022
cf3f3e6
Reinstate lost object wrapper
DaveCTurner Dec 7, 2022
e334219
Missing null check
DaveCTurner Dec 7, 2022
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
14 changes: 6 additions & 8 deletions ...c/main/java/org/elasticsearch/action/admin/cluster/node/tasks/list/ListTasksResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,10 +32,8 @@
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static org.elasticsearch.xcontent.ConstructingObjectParser.optionalConstructorArg;

Expand Down Expand Up @@ -166,9 +164,9 @@ public ChunkedToXContent groupedByNode(Supplier<DiscoveryNodes> nodesInCluster)
toXContentCommon(builder, params);
builder.startObject("nodes");
return builder;
}), getPerNodeTasks().entrySet().stream().flatMap(entry -> {
}), Iterators.flatMap(getPerNodeTasks().entrySet().iterator(), entry -> {
DiscoveryNode node = discoveryNodes.get(entry.getKey());
return Stream.<Stream<ToXContent>>of(Stream.of((builder, params) -> {
return Iterators.concat(Iterators.single((builder, params) -> {
builder.startObject(entry.getKey());
if (node != null) {
// If the node is no longer part of the cluster, oh well, we'll just skip its useful information.
Expand All @@ -193,17 +191,17 @@ public ChunkedToXContent groupedByNode(Supplier<DiscoveryNodes> nodesInCluster)
}
builder.startObject(TASKS);
return builder;
}), entry.getValue().stream().<ToXContent>map(task -> (builder, params) -> {
}), Iterators.flatMap(entry.getValue().iterator(), task -> Iterators.<ToXContent>single((builder, params) -> {
builder.startObject(task.taskId().toString());
task.toXContent(builder, params);
builder.endObject();
return builder;
}), Stream.of((builder, params) -> {
})), Iterators.<ToXContent>single((builder, params) -> {
builder.endObject();
builder.endObject();
return builder;
})).flatMap(Function.identity());
}).iterator(), Iterators.single((builder, params) -> {
}));
}), Iterators.<ToXContent>single((builder, params) -> {
builder.endObject();
builder.endObject();
return builder;
Expand Down
13 changes: 3 additions & 10 deletions ...java/org/elasticsearch/action/admin/cluster/snapshots/status/SnapshotsStatusResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,9 +22,6 @@
import java.util.Iterator;
import java.util.List;
import java.util.Objects;
import java.util.Spliterator;
import java.util.Spliterators;
import java.util.stream.StreamSupport;

import static org.elasticsearch.xcontent.ConstructingObjectParser.constructorArg;

Expand Down Expand Up @@ -90,13 +87,9 @@ public int hashCode() {

@Override
public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params params) {
return Iterators.concat(
Iterators.single((ToXContent) (b, p) -> b.startObject().startArray("snapshots")),
snapshots.stream()
.flatMap(
s -> StreamSupport.stream(Spliterators.spliteratorUnknownSize(s.toXContentChunked(params), Spliterator.ORDERED), false)
)
.iterator(),
return Iterators.<ToXContent>concat(
Iterators.single((b, p) -> b.startObject().startArray("snapshots")),
Iterators.flatMap(snapshots.iterator(), s -> s.toXContentChunked(params)),
Iterators.single((b, p) -> b.endArray().endObject())
);
}
Expand Down
2 changes: 1 addition & 1 deletion server/src/main/java/org/elasticsearch/cluster/metadata/IndexGraveyard.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,7 +127,7 @@ public boolean containsIndex(final Index index) {

@Override
public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params ignored) {
return ChunkedToXContentHelper.array(TOMBSTONES_FIELD.getPreferredName(), tombstones);
return ChunkedToXContentHelper.array(TOMBSTONES_FIELD.getPreferredName(), tombstones.iterator());
}

public static IndexGraveyard fromXContent(final XContentParser parser) throws IOException {
Expand Down
22 changes: 7 additions & 15 deletions server/src/main/java/org/elasticsearch/cluster/metadata/Metadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,16 +75,13 @@
import java.util.Optional;
import java.util.Set;
import java.util.SortedMap;
import java.util.Spliterator;
import java.util.Spliterators;
import java.util.TreeMap;
import java.util.function.BiPredicate;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

import static org.elasticsearch.cluster.metadata.LifecycleExecutionState.ILM_CUSTOM_METADATA_KEY;
import static org.elasticsearch.common.settings.Settings.readSettingsFromStream;
Expand Down Expand Up @@ -1340,7 +1337,7 @@ public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params p) {
? ChunkedToXContentHelper.wrapWithObject("indices", indices().values().iterator())
: Collections.emptyIterator();

return Iterators.concat(start, Iterators.single((builder, params) -> {
return Iterators.concat(start, Iterators.<ToXContent>single((builder, params) -> {
builder.field("cluster_uuid", clusterUUID);
builder.field("cluster_uuid_committed", clusterUUIDCommitted);
builder.startObject("cluster_coordination");
Expand All @@ -1362,17 +1359,12 @@ public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params p) {
.iterator()
),
indices,
customs().entrySet()
.stream()
.filter(cursor -> cursor.getValue().context().contains(context))
.map(
cursor -> (ChunkedToXContent) params -> ChunkedToXContentHelper.wrapWithObject(
cursor.getKey(),
cursor.getValue().toXContentChunked(params)
)
)
.flatMap(s -> StreamSupport.stream(Spliterators.spliteratorUnknownSize(s.toXContentChunked(p), Spliterator.ORDERED), false))
.iterator(),
Iterators.flatMap(
customs.entrySet().iterator(),
entry -> entry.getValue().context().contains(context)
? ChunkedToXContentHelper.wrapWithObject(entry.getKey(), entry.getValue().toXContentChunked(p))
: Collections.emptyIterator()
),
ChunkedToXContentHelper.wrapWithObject("reserved_state", reservedStateMetadata().values().iterator()),
ChunkedToXContentHelper.endObject()
);
Expand Down
91 changes: 73 additions & 18 deletions server/src/main/java/org/elasticsearch/common/collect/Iterators.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,9 +8,13 @@

package org.elasticsearch.common.collect;

import org.elasticsearch.core.Nullable;

import java.util.Collections;
import java.util.Iterator;
import java.util.NoSuchElementException;
import java.util.Objects;
import java.util.function.Function;

public class Iterators {

Expand Down Expand Up @@ -43,44 +47,45 @@ public static <T> Iterator<T> concat(Iterator<? extends T>... iterators) {
throw new NullPointerException("iterators");
}

// explicit generic type argument needed for type inference
return new ConcatenatedIterator<T>(iterators);
for (int i = 0; i < iterators.length; i++) {
if (iterators[i].hasNext()) {
// explicit generic type argument needed for type inference
return new ConcatenatedIterator<T>(iterators, i);
}
}

return Collections.emptyIterator();
}

static class ConcatenatedIterator<T> implements Iterator<T> {
private static class ConcatenatedIterator<T> implements Iterator<T> {
private final Iterator<? extends T>[] iterators;
private int index = 0;
private int index;

@SafeVarargs
@SuppressWarnings("varargs")
ConcatenatedIterator(Iterator<? extends T>... iterators) {
if (iterators == null) {
throw new NullPointerException("iterators");
}
for (int i = 0; i < iterators.length; i++) {
ConcatenatedIterator(Iterator<? extends T>[] iterators, int startIndex) {
for (int i = startIndex; i < iterators.length; i++) {
if (iterators[i] == null) {
throw new NullPointerException("iterators[" + i + "]");
}
}
this.iterators = iterators;
this.index = startIndex;
}

@Override
public boolean hasNext() {
boolean hasNext = false;
while (index < iterators.length && (hasNext = iterators[index].hasNext()) == false) {
index++;
Copy link
Contributor Author

Choose a reason for hiding this comment

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

I find it weird to be changing state in hasNext() like this, I prefer to do all the moving in next() so that we're always pointing at a nonempty iterator until done.

Copy link
Member

Choose a reason for hiding this comment

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

++

}

return hasNext;
return index < iterators.length;
}

@Override
public T next() {
if (hasNext() == false) {
throw new NoSuchElementException();
}
return iterators[index].next();
final T value = iterators[index].next();
while (index < iterators.length && iterators[index].hasNext() == false) {
index++;
}
return value;
}
}

Expand Down Expand Up @@ -110,4 +115,54 @@ public T next() {
return array[index++];
}
}

public static <T, U> Iterator<? extends U> flatMap(Iterator<? extends T> input, Function<T, Iterator<? extends U>> fn) {
while (input.hasNext()) {
final var value = fn.apply(input.next());
if (value.hasNext()) {
return new FlatMapIterator<>(input, fn, value);
}
}

return Collections.emptyIterator();
}

private static final class FlatMapIterator<T, U> implements Iterator<U> {

private final Iterator<? extends T> input;
private final Function<T, Iterator<? extends U>> fn;

@Nullable // if finished, otherwise currentOutput.hasNext() is true
private Iterator<? extends U> currentOutput;

FlatMapIterator(Iterator<? extends T> input, Function<T, Iterator<? extends U>> fn, Iterator<? extends U> firstOutput) {
this.input = input;
this.fn = fn;
this.currentOutput = firstOutput;
}

@Override
public boolean hasNext() {
return currentOutput != null;
}

@Override
public U next() {
if (hasNext() == false) {
throw new NoSuchElementException();
}
// noinspection ConstantConditions this is for documentation purposes
assert currentOutput != null && currentOutput.hasNext();
final U value = currentOutput.next();
while (currentOutput != null && currentOutput.hasNext() == false) {
if (input.hasNext()) {
currentOutput = fn.apply(input.next());
} else {
currentOutput = null;
}
}
return value;
}
}

}
4 changes: 2 additions & 2 deletions server/src/main/java/org/elasticsearch/common/xcontent/ChunkedToXContentHelper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,8 +62,8 @@ public static Iterator<ToXContent> field(String name, boolean value) {
return Iterators.single(((builder, params) -> builder.field(name, value)));
}

public static Iterator<ToXContent> array(String name, Iterable<? extends ToXContent> iterable) {
return Iterators.concat(ChunkedToXContentHelper.startArray(name), iterable.iterator(), ChunkedToXContentHelper.endArray());
public static Iterator<ToXContent> array(String name, Iterator<? extends ToXContent> contents) {
return Iterators.concat(ChunkedToXContentHelper.startArray(name), contents, ChunkedToXContentHelper.endArray());
}

public static <T extends ToXContent> Iterator<ToXContent> wrapWithObject(String name, Iterator<T> iterator) {
Expand Down
27 changes: 8 additions & 19 deletions server/src/main/java/org/elasticsearch/health/Diagnosis.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.common.collect.Iterators;
import org.elasticsearch.common.xcontent.ChunkedToXContent;
import org.elasticsearch.common.xcontent.ChunkedToXContentHelper;
import org.elasticsearch.core.Nullable;
import org.elasticsearch.xcontent.ToXContent;

Expand All @@ -19,9 +20,6 @@
import java.util.Iterator;
import java.util.List;
import java.util.Objects;
import java.util.Spliterator;
import java.util.Spliterators;
import java.util.stream.StreamSupport;

import static org.elasticsearch.health.HealthService.HEALTH_API_ID_PREFIX;

Expand Down Expand Up @@ -78,7 +76,7 @@ public Resource(Collection<DiscoveryNode> nodes) {

@Override
public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params outerParams) {
Iterator<? extends ToXContent> valuesIterator;
final Iterator<? extends ToXContent> valuesIterator;
if (nodes != null) {
valuesIterator = nodes.stream().map(node -> (ToXContent) (builder, params) -> {
builder.startObject();
Expand All @@ -92,11 +90,7 @@ public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params outerP
} else {
valuesIterator = values.stream().map(value -> (ToXContent) (builder, params) -> builder.value(value)).iterator();
}
return Iterators.concat(
Iterators.single((ToXContent) (builder, params) -> builder.startArray(type.displayValue)),
valuesIterator,
Iterators.single((builder, params) -> builder.endArray())
);
return ChunkedToXContentHelper.array(type.displayValue, valuesIterator);
}

@Override
Expand Down Expand Up @@ -148,16 +142,11 @@ public String getUniqueId() {

@Override
public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params outerParams) {
Iterator<? extends ToXContent> resourcesIterator = Collections.emptyIterator();
if (affectedResources != null && affectedResources.size() > 0) {
resourcesIterator = affectedResources.stream()
.flatMap(
s -> StreamSupport.stream(
Spliterators.spliteratorUnknownSize(s.toXContentChunked(outerParams), Spliterator.ORDERED),
false
)
)
.iterator();
final Iterator<? extends ToXContent> resourcesIterator;
if (affectedResources == null) {
resourcesIterator = Collections.emptyIterator();
} else {
resourcesIterator = Iterators.flatMap(affectedResources.iterator(), s -> s.toXContentChunked(outerParams));
}
return Iterators.concat(Iterators.single((ToXContent) (builder, params) -> {
builder.startObject();
Expand Down
18 changes: 5 additions & 13 deletions server/src/main/java/org/elasticsearch/health/HealthIndicatorResult.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,9 +15,6 @@
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Spliterator;
import java.util.Spliterators;
import java.util.stream.StreamSupport;

public record HealthIndicatorResult(
String name,
Expand All @@ -29,16 +26,11 @@ public record HealthIndicatorResult(
) implements ChunkedToXContent {
@Override
public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params outerParams) {
Iterator<? extends ToXContent> diagnosisIterator = Collections.emptyIterator();
if (diagnosisList != null && diagnosisList.isEmpty() == false) {
diagnosisIterator = diagnosisList.stream()
.flatMap(
s -> StreamSupport.stream(
Spliterators.spliteratorUnknownSize(s.toXContentChunked(outerParams), Spliterator.ORDERED),
false
)
)
.iterator();
final Iterator<? extends ToXContent> diagnosisIterator;
if (diagnosisList == null) {
diagnosisIterator = Collections.emptyIterator();
} else {
diagnosisIterator = Iterators.flatMap(diagnosisList.iterator(), s -> s.toXContentChunked(outerParams));
}
return Iterators.concat(Iterators.single((ToXContent) (builder, params) -> {
builder.startObject();
Expand Down
2 changes: 1 addition & 1 deletion server/src/main/java/org/elasticsearch/ingest/IngestMetadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,7 +98,7 @@ public static IngestMetadata fromXContent(XContentParser parser) throws IOExcept

@Override
public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params ignored) {
return ChunkedToXContentHelper.array(PIPELINES_FIELD.getPreferredName(), pipelines.values());
return ChunkedToXContentHelper.array(PIPELINES_FIELD.getPreferredName(), pipelines.values().iterator());
}

@Override
Expand Down
2 changes: 1 addition & 1 deletion server/src/main/java/org/elasticsearch/persistent/PersistentTasksCustomMetadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -558,7 +558,7 @@ public static NamedDiff<Metadata.Custom> readDiffFrom(StreamInput in) throws IOE
public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params ignored) {
return Iterators.concat(
Iterators.single((builder, params) -> builder.field("last_allocation_id", lastAllocationId)),
ChunkedToXContentHelper.array("tasks", tasks.values())
ChunkedToXContentHelper.array("tasks", tasks.values().iterator())
);
}

Expand Down
Loading