Understanding downstream reduction implementation

Question

I'm trying to understand the implementation of downstream reduction in JDK. Here is it:

   public static <T, K, D, A, M extends Map<K, D>>
    Collector<T, ?, M> groupingBy(Function<? super T, ? extends K> classifier,
                                  Supplier<M> mapFactory,
                                  Collector<? super T, A, D> downstream) {
        Supplier<A> downstreamSupplier = downstream.supplier();
        BiConsumer<A, ? super T> downstreamAccumulator = downstream.accumulator();
        BiConsumer<Map<K, A>, T> accumulator = (m, t) -> {
            K key = Objects.requireNonNull(classifier.apply(t), "element cannot be mapped to a null key");
            A container = m.computeIfAbsent(key, k -> downstreamSupplier.get());
            downstreamAccumulator.accept(container, t);
        };
        BinaryOperator<Map<K, A>> merger = Collectors.<K, A, Map<K, A>>mapMerger(downstream.combiner());
        @SuppressWarnings("unchecked")
        Supplier<Map<K, A>> mangledFactory = (Supplier<Map<K, A>>) mapFactory;

        if (downstream.characteristics().contains(Collector.Characteristics.IDENTITY_FINISH)) {
            return new CollectorImpl<>(mangledFactory, accumulator, merger, CH_ID);
        }
        else {
            @SuppressWarnings("unchecked")
            Function<A, A> downstreamFinisher = 
                       (Function<A, A>) downstream.finisher();  //1, <------------- HERE
            Function<Map<K, A>, M> finisher = intermediate -> {
                intermediate.replaceAll((k, v) -> downstreamFinisher.apply(v));
                @SuppressWarnings("unchecked")
                M castResult = (M) intermediate;
                return castResult;
            };
            return new CollectorImpl<>(mangledFactory, accumulator, merger, finisher, CH_NOID);
        }
    }

At //1, the downstreamFinisher is of type Function<A, D>. Judging by the type parameters declarations <T, K, D, A, M extends Map<K, D>>, the type D does not depend on A. So why do we cast it to Function<A, A>. I think, the type D may not even be a subclass of A.

What did I miss?

Answer 1

This collector is actually violating the generic type safety of the Map, if the downstream collector doesn’t have an identity finisher and the finisher function returns a different type than the intermediate container type.

During the collect operation, the map will hold objects of type A, i.e. the intermediate container type. Then, at the end of the operation, groupingBy’s finisher will go through the map and apply the finisher function to each value and replace it with the final result.

Of course, this can’t be implemented without unchecked operations. There are multiple ways to do it, the variant you have posted, changes the type of the map supplier from Supplier<M> to Supplier<Map<K, A>> (the first unchecked operation), so the compiler accepts that the map will hold values of type A rather than D. That’s why the finisher function must be changed to Function<A,A> (the second unchecked operation), so it can be used in the map’s replaceAll operation which appears to require objects of type A despite it’s actually D. Finally, the result map must be cast to M (the third unchecked operation) to get an object of the expected result type M, which the supplier actually supplied.

The correct type safe alternative would be to use different maps and perform the finishing operation by populating the result map with the result of converting the values of the intermediate map. Not only might this be an expensive operation, it would require a second supplier for the intermediate map, as the provided supplier only produces maps suitable for the final result. So apparently, the developers decided this to be an acceptable breach of the type safety.

Note that you can notice the unsafe operation, when you try to use a Map implementation which enforces type safety:

Stream.of("foo", "bar").collect(Collectors.groupingBy(String::length,
    () -> Collections.checkedMap(new HashMap<>(), Integer.class, Long.class),
    Collectors.counting()));

will produce a ClassCastException with this implementation because it tries to put an intermediate container (an array) instead of the Long into the Map.