java - resueltos - Partición de un flujo por una función discriminadora

potencia electrica formula y ejemplos

Question

Una de las características que faltan en la API de Streams es la transformación "partición por", por ejemplo, como se define en Clojure . Digamos que quiero reproducir la combinación de búsqueda de Hibernate: quiero emitir una sola instrucción SELECT de SQL para recibir este tipo de objetos del resultado:

class Family {
String surname;
List<String> members;
}

Yo emito:

SELECT f.name, m.name
FROM Family f JOIN Member m on m.family_id = f.id
ORDER BY f.name

y recupero un flujo plano de (f.name, m.name) registros. Ahora necesito transformarlo en un flujo de objetos de la Family , con una lista de sus miembros dentro. Supongamos que ya tengo un Stream<ResultRow> ; ahora necesito transformarlo en una Stream<List<ResultRow>> y luego actuar sobre eso con una transformación de mapeo que la convierte en una Stream<Family> .

La semántica de la transformación es la siguiente: siga recopilando el flujo en una List mientras la función discriminadora provista siga devolviendo el mismo valor; tan pronto como el valor cambie, emita la List como un elemento del flujo de salida y comience a recopilar una nueva List .

Espero poder escribir este tipo de código (ya tengo el método resultStream ):

Stream<ResultRow> dbStream = resultStream(queryBuilder.createQuery(
"SELECT f.name, m.name"
+ " FROM Family f JOIN Member m on m.family_id = f.id"
+ " ORDER BY f.name"));
Stream<List<ResultRow> partitioned = partitionBy(r -> r.string(0), dbStream);
Stream<Family> = partitioned.map(rs -> {
Family f = new Family(rs.get(0).string(0));
f.members = rs.stream().map(r -> r.string(1)).collect(toList());
return f;
});

No hace falta decir que espero que el flujo resultante permanezca perezoso (sin materializar) ya que quiero poder procesar un conjunto de resultados de cualquier tamaño sin alcanzar ningún límite de memoria O (n). Sin este requisito crucial, me sentiría feliz con la groupingBy proporcionada por coleccionista.

Answer 1

La solución requiere que definamos un Spliterator personalizado que se pueda usar para construir el flujo particionado. Tendremos que acceder al flujo de entrada a través de su propio separador y envolverlo en el nuestro. El flujo de salida se construye a partir de nuestro separador personalizado.

El siguiente Spliterator convertirá cualquier Stream<E> en un Stream<List<E>> provisto de una Function<E, ?> Como la función discriminadora. Tenga en cuenta que la secuencia de entrada debe estar ordenada para que esta operación tenga sentido.

public class PartitionBySpliterator<E> extends AbstractSpliterator<List<E>> {
private final Spliterator<E> spliterator;
private final Function<? super E, ?> partitionBy;
private HoldingConsumer<E> holder;
private Comparator<List<E>> comparator;
public PartitionBySpliterator(Spliterator<E> toWrap, Function<? super E, ?> partitionBy) {
super(Long.MAX_VALUE, toWrap.characteristics() & ~SIZED | NONNULL);
this.spliterator = toWrap;
this.partitionBy = partitionBy;
}
public static <E> Stream<List<E>> partitionBy(Function<E, ?> partitionBy, Stream<E> in) {
return StreamSupport.stream(new PartitionBySpliterator<>(in.spliterator(), partitionBy), false);
}
@Override public boolean tryAdvance(Consumer<? super List<E>> action) {
final HoldingConsumer<E> h;
if (holder == null) {
h = new HoldingConsumer<>();
if (!spliterator.tryAdvance(h)) return false;
holder = h;
}
else h = holder;
final ArrayList<E> partition = new ArrayList<>();
final Object partitionKey = partitionBy.apply(h.value);
boolean didAdvance;
do partition.add(h.value);
while ((didAdvance = spliterator.tryAdvance(h))
&& Objects.equals(partitionBy.apply(h.value), partitionKey));
if (!didAdvance) holder = null;
action.accept(partition);
return true;
}
static final class HoldingConsumer<T> implements Consumer<T> {
T value;
@Override public void accept(T value) { this.value = value; }
}
@Override public Comparator<? super List<E>> getComparator() {
final Comparator<List<E>> c = this.comparator;
return c != null? c : (this.comparator = comparator());
}
private Comparator<List<E>> comparator() {
@SuppressWarnings({"unchecked","rawtypes"})
final Comparator<? super E> innerComparator =
Optional.ofNullable(spliterator.getComparator())
.orElse((Comparator) naturalOrder());
return (left, right) -> {
final int c = innerComparator.compare(left.get(0), right.get(0));
return c != 0? c : innerComparator.compare(
left.get(left.size() - 1), right.get(right.size() - 1));
};
}
}

Answer 2

Para aquellos de ustedes que solo quieren particionar un flujo, hay mapeadores y recolectores para eso.

class Person {
String surname;
String forename;
public Person(String surname, String forename) {
this.surname = surname;
this.forename = forename;
}
@Override
public String toString() {
return forename;
}
}
class Family {
String surname;
List<Person> members;
public Family(String surname, List<Person> members) {
this.surname = surname;
this.members = members;
}
@Override
public String toString() {
return "Family{" + "surname=" + surname + ", members=" + members + ''}'';
}
}
private void test() {
String[][] data = {
{"Kray", "Ronald"},
{"Kray", "Reginald"},
{"Dors", "Diana"},};
// Their families.
Stream<Family> families = Arrays.stream(data)
// Build people
.map(a -> new Person(a[0], a[1]))
// Collect into a Map<String,List<Person>> as families
.collect(Collectors.groupingBy(p -> p.surname))
// Convert them to families.
.entrySet().stream()
.map(p -> new Family(p.getKey(), p.getValue()));
families.forEach(f -> System.out.println(f));
}

Answer 3

Se puede hacer collapse con StreamEx

StreamEx.of(queryBuilder.createQuery(
"SELECT f.name, m.name"
+ " FROM Family f JOIN Member m on m.family_id = f.id"
+ " ORDER BY f.name"))
.collapse((a, b) -> a.string(0).equals(b.string(0)), Collectors.toList())
.map(l -> new Family(l.get(0).string(0), StreamEx.of(l).map(r -> r.string(1)).toList()))
.forEach(System.out::println);