content: /*
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you under the Apache License, Version 2.0 (the
   * "License"); you may not use this file except in compliance
   * with the License.  You may obtain a copy of the License at
   *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.beam.runners.spark.structuredstreaming.translation.batch;
  
  import static org.apache.beam.runners.spark.structuredstreaming.translation.helpers.EncoderHelpers.collectionEncoder;
  import static org.apache.beam.runners.spark.structuredstreaming.translation.helpers.EncoderHelpers.encoderOf;
  import static org.apache.beam.runners.spark.structuredstreaming.translation.helpers.EncoderHelpers.mapEncoder;
  import static org.apache.beam.runners.spark.structuredstreaming.translation.helpers.EncoderHelpers.mutablePairEncoder;
  import static org.apache.beam.sdk.transforms.windowing.PaneInfo.NO_FIRING;
  import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument;
  import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterators.peekingIterator;
  
  import java.util.Collection;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.Map;
  import java.util.Map.Entry;
  import java.util.Set;
  import java.util.TreeMap;
  import java.util.function.BiFunction;
  import java.util.function.BinaryOperator;
  import java.util.function.Function;
  import org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.Fun1;
  import org.apache.beam.sdk.annotations.Internal;
  import org.apache.beam.sdk.transforms.Combine.CombineFn;
  import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
  import org.apache.beam.sdk.transforms.windowing.IntervalWindow;
  import org.apache.beam.sdk.transforms.windowing.Sessions;
  import org.apache.beam.sdk.transforms.windowing.TimestampCombiner;
  import org.apache.beam.sdk.transforms.windowing.WindowFn;
  import org.apache.beam.sdk.values.WindowedValue;
  import org.apache.beam.sdk.values.WindowedValues;
  import org.apache.beam.sdk.values.WindowingStrategy;
  import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Collections2;
  import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
  import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableSet;
  import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.PeekingIterator;
  import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Sets;
  import org.apache.spark.sql.Encoder;
  import org.apache.spark.sql.expressions.Aggregator;
  import org.apache.spark.util.MutablePair;
  import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
  import org.checkerframework.checker.nullness.qual.NonNull;
  import org.checkerframework.checker.nullness.qual.Nullable;
  import org.checkerframework.checker.nullness.qual.PolyNull;
  import org.joda.time.Instant;
  
  @Internal
  class Aggregators {
  
    /**
     * Creates simple value {@link Aggregator} that is not window aware.
     *
     * @param <ValT> {@link CombineFn} input type
     * @param <AccT> {@link CombineFn} accumulator type
     * @param <ResT> {@link CombineFn} / {@link Aggregator} result type
     * @param <InT> {@link Aggregator} input type
     */
    static <ValT, AccT, ResT, InT> Aggregator<InT, ?, ResT> value(
        CombineFn<ValT, AccT, ResT> fn,
        Fun1<InT, ValT> valueFn,
        Encoder<AccT> accEnc,
        Encoder<ResT> outEnc) {
      return new ValueAggregator<>(fn, valueFn, accEnc, outEnc);
    }
  
    /**
     * Creates windowed Spark {@link Aggregator} depending on the provided Beam {@link WindowFn}s.
     *
     * <p>Specialised implementations are provided for:
     * <li>{@link Sessions}
     * <li>Non merging window functions
     * <li>Merging window functions
     *
     * @param <ValT> {@link CombineFn} input type
     * @param <AccT> {@link CombineFn} accumulator type
     * @param <ResT> {@link CombineFn} / {@link Aggregator} result type
     * @param <InT> {@link Aggregator} input type
     */
    static <ValT, AccT, ResT, InT>
        Aggregator<WindowedValue<InT>, ?, Collection<WindowedValue<ResT>>> windowedValue(
            CombineFn<ValT, AccT, ResT> fn,
            Fun1<WindowedValue<InT>, ValT> valueFn,
            WindowingStrategy<?, ?> windowing,
            Encoder<BoundedWindow> windowEnc,
            Encoder<AccT> accEnc,
            Encoder<WindowedValue<ResT>> outEnc) {
      if (!windowing.needsMerge()) {
        return new NonMergingWindowedAggregator<>(fn, valueFn, windowing, windowEnc, accEnc, outEnc);
      } else if (windowing.getWindowFn().getClass().equals(Sessions.class)) {
        return new SessionsAggregator<>(fn, valueFn, windowing, (Encoder) windowEnc, accEnc, outEnc);
      }
      return new MergingWindowedAggregator<>(fn, valueFn, windowing, windowEnc, accEnc, outEnc);
    }
  
    /**
     * Simple value {@link Aggregator} that is not window aware.
     *
     * @param <ValT> {@link CombineFn} input type
     * @param <AccT> {@link CombineFn} accumulator type
     * @param <ResT> {@link CombineFn} / {@link Aggregator} result type
     * @param <InT> {@link Aggregator} input type
     */
    private static class ValueAggregator<ValT, AccT, ResT, InT>
        extends CombineFnAggregator<ValT, AccT, ResT, InT, AccT, ResT> {
  
      public ValueAggregator(
          CombineFn<ValT, AccT, ResT> fn,
          Fun1<InT, ValT> valueFn,
          Encoder<AccT> accEnc,
          Encoder<ResT> outEnc) {
        super(fn, valueFn, accEnc, outEnc);
      }
  
      @Override
      public AccT zero() {
        return emptyAcc();
      }
  
      @Override
      public AccT reduce(AccT buff, InT in) {
        return addToAcc(buff, value(in));
      }
  
      @Override
      public AccT merge(AccT b1, AccT b2) {
        return mergeAccs(b1, b2);
      }
  
      @Override
      public ResT finish(AccT buff) {
        return extract(buff);
      }
    }
  
    /**
     * Specialized windowed Spark {@link Aggregator} for Beam {@link WindowFn}s of type {@link
     * Sessions}. The aggregator uses a {@link TreeMap} as buffer to maintain ordering of the {@link
     * IntervalWindow}s and merge these more efficiently.
     *
     * <p>For efficiency, this aggregator re-implements {@link
     * Sessions#mergeWindows(WindowFn.MergeContext)} to leverage the already sorted buffer.
     *
     * @param <ValT> {@link CombineFn} input type
     * @param <AccT> {@link CombineFn} accumulator type
     * @param <ResT> {@link CombineFn} / {@link Aggregator} result type
     * @param <InT> {@link Aggregator} input type
     */
    private static class SessionsAggregator<ValT, AccT, ResT, InT>
        extends WindowedAggregator<
            ValT,
            AccT,
            ResT,
            InT,
            IntervalWindow,
            TreeMap<IntervalWindow, MutablePair<Instant, AccT>>> {
  
      SessionsAggregator(
          CombineFn<ValT, AccT, ResT> combineFn,
          Fun1<WindowedValue<InT>, ValT> valueFn,
          WindowingStrategy<?, ?> windowing,
          Encoder<IntervalWindow> windowEnc,
          Encoder<AccT> accEnc,
          Encoder<WindowedValue<ResT>> outEnc) {
        super(combineFn, valueFn, windowing, windowEnc, accEnc, outEnc, (Class) TreeMap.class);
        checkArgument(windowing.getWindowFn().getClass().equals(Sessions.class));
      }
  
      @Override
      public final TreeMap<IntervalWindow, MutablePair<Instant, AccT>> zero() {
        return new TreeMap<>();
      }
  
      @Override
      @SuppressWarnings("keyfor")
      public TreeMap<IntervalWindow, MutablePair<Instant, AccT>> reduce(
          TreeMap<IntervalWindow, MutablePair<Instant, AccT>> buff, WindowedValue<InT> input) {
        for (IntervalWindow window : (Collection<IntervalWindow>) input.getWindows()) {
          @MonotonicNonNull MutablePair<Instant, AccT> acc = null;
          @MonotonicNonNull IntervalWindow first = null, last = null;
          // start with window before or equal to new window (if exists)
          @Nullable Entry<IntervalWindow, MutablePair<Instant, AccT>> lower = buff.floorEntry(window);
          if (lower != null && window.intersects(lower.getKey())) {
            // if intersecting, init accumulator and extend window to span both
            acc = lower.getValue();
            window = window.span(lower.getKey());
            first = last = lower.getKey();
          }
          // merge following windows in order if they intersect, then stop
          for (Entry<IntervalWindow, MutablePair<Instant, AccT>> entry :
              buff.tailMap(window, false).entrySet()) {
            MutablePair<Instant, AccT> entryAcc = entry.getValue();
            IntervalWindow entryWindow = entry.getKey();
            if (window.intersects(entryWindow)) {
              // extend window and merge accumulators
              window = window.span(entryWindow);
              acc = acc == null ? entryAcc : mergeAccs(window, acc, entryAcc);
              if (first == null) {
                // there was no previous (lower) window intersecting the input window
                first = last = entryWindow;
              } else {
                last = entryWindow;
              }
            } else {
              break; // stop, later windows won't intersect either
            }
          }
          if (first != null && last != null) {
            // remove entire subset from from first to last after it got merged into acc
            buff.navigableKeySet().subSet(first, true, last, true).clear();
          }
          // add input and get accumulator for new (potentially merged) window
          buff.put(window, addToAcc(window, acc, value(input), input.getTimestamp()));
        }
        return buff;
      }
  
      @Override
      public TreeMap<IntervalWindow, MutablePair<Instant, AccT>> merge(
          TreeMap<IntervalWindow, MutablePair<Instant, AccT>> b1,
          TreeMap<IntervalWindow, MutablePair<Instant, AccT>> b2) {
        if (b1.isEmpty()) {
          return b2;
        } else if (b2.isEmpty()) {
          return b1;
        }
        // Init new tree map to merge both buffers
        TreeMap<IntervalWindow, MutablePair<Instant, AccT>> res = zero();
        PeekingIterator<Entry<IntervalWindow, MutablePair<Instant, AccT>>> it1 =
            peekingIterator(b1.entrySet().iterator());
        PeekingIterator<Entry<IntervalWindow, MutablePair<Instant, AccT>>> it2 =
            peekingIterator(b2.entrySet().iterator());
  
        @Nullable MutablePair<Instant, AccT> acc = null;
        @Nullable IntervalWindow window = null;
        while (it1.hasNext() || it2.hasNext()) {
          // pick iterator with the smallest window ahead and forward it
          Entry<IntervalWindow, MutablePair<Instant, AccT>> nextMin =
              (it1.hasNext() && it2.hasNext())
                  ? it1.peek().getKey().compareTo(it2.peek().getKey()) <= 0 ? it1.next() : it2.next()
                  : it1.hasNext() ? it1.next() : it2.next();
          if (window != null && window.intersects(nextMin.getKey())) {
            // extend window and merge accumulators if intersecting
            window = window.span(nextMin.getKey());
            acc = mergeAccs(window, acc, nextMin.getValue());
          } else {
            // store window / accumulator if necessary and continue with next minimum
            if (window != null && acc != null) {
              res.put(window, acc);
            }
            acc = nextMin.getValue();
            window = nextMin.getKey();
          }
        }
        if (window != null && acc != null) {
          res.put(window, acc);
        }
        return res;
      }
    }
  
    /**
     * Merging windowed Spark {@link Aggregator} using a Map of {@link BoundedWindow}s as aggregation
     * buffer. When reducing new input, a windowed accumulator is created for each new window of the
     * input that doesn't overlap with existing windows. Otherwise, if the window is known or
     * overlaps, the window is extended accordingly and accumulators are merged.
     *
     * @param <ValT> {@link CombineFn} input type
     * @param <AccT> {@link CombineFn} accumulator type
     * @param <ResT> {@link CombineFn} / {@link Aggregator} result type
     * @param <InT> {@link Aggregator} input type
     */
    private static class MergingWindowedAggregator<ValT, AccT, ResT, InT>
        extends NonMergingWindowedAggregator<ValT, AccT, ResT, InT> {
  
      private final WindowFn<ValT, BoundedWindow> windowFn;
  
      public MergingWindowedAggregator(
          CombineFn<ValT, AccT, ResT> combineFn,
          Fun1<WindowedValue<InT>, ValT> valueFn,
          WindowingStrategy<?, ?> windowing,
          Encoder<BoundedWindow> windowEnc,
          Encoder<AccT> accEnc,
          Encoder<WindowedValue<ResT>> outEnc) {
        super(combineFn, valueFn, windowing, windowEnc, accEnc, outEnc);
        windowFn = (WindowFn<ValT, BoundedWindow>) windowing.getWindowFn();
      }
  
      @Override
      protected Map<BoundedWindow, MutablePair<Instant, AccT>> reduce(
          Map<BoundedWindow, MutablePair<Instant, AccT>> buff,
          Collection<BoundedWindow> windows,
          ValT value,
          Instant timestamp) {
        if (buff.isEmpty()) {
          // no windows yet to be merged, use the non-merging behavior of super
          return super.reduce(buff, windows, value, timestamp);
        }
        // Merge multiple windows into one target window using the reducer function if the window
        // already exists. Otherwise, the input value is added to the accumulator. Merged windows are
        // removed from the accumulator map.
        Function<BoundedWindow, ReduceFn<AccT>> accFn =
            target ->
                (acc, w) -> {
                  MutablePair<Instant, AccT> accW = buff.remove(w);
                  return (accW != null)
                      ? mergeAccs(w, acc, accW)
                      : addToAcc(w, acc, value, timestamp);
                };
        Set<BoundedWindow> unmerged = mergeWindows(buff, ImmutableSet.copyOf(windows), accFn);
        if (!unmerged.isEmpty()) {
          // remaining windows don't have to be merged
          return super.reduce(buff, unmerged, value, timestamp);
        }
        return buff;
      }
  
      @Override
      public Map<BoundedWindow, MutablePair<Instant, AccT>> merge(
          Map<BoundedWindow, MutablePair<Instant, AccT>> b1,
          Map<BoundedWindow, MutablePair<Instant, AccT>> b2) {
        // Merge multiple windows into one target window using the reducer function. Merged windows
        // are removed from both accumulator maps
        Function<BoundedWindow, ReduceFn<AccT>> reduceFn =
            target -> (acc, w) -> mergeAccs(w, mergeAccs(w, acc, b1.remove(w)), b2.remove(w));
  
        Set<BoundedWindow> unmerged = b2.keySet();
        unmerged = mergeWindows(b1, unmerged, reduceFn);
        if (!unmerged.isEmpty()) {
          // keep only unmerged windows in 2nd accumulator map, continue using "non-merging" merge
          b2.keySet().retainAll(unmerged);
          return super.merge(b1, b2);
        }
        return b1;
      }
  
      /** Reduce function to merge multiple windowed accumulator values into one target window. */
      private interface ReduceFn<AccT>
          extends BiFunction<
              @Nullable MutablePair<Instant, AccT>,
              BoundedWindow,
              @Nullable MutablePair<Instant, AccT>> {}
  
      /**
       * Attempt to merge windows of accumulator map with additional windows using the reducer
       * function. The reducer function must support {@code null} as zero value.
       *
       * @return The subset of additional windows that don't require a merge.
       */
      private Set<BoundedWindow> mergeWindows(
          Map<BoundedWindow, MutablePair<Instant, AccT>> buff,
          Set<BoundedWindow> newWindows,
          Function<BoundedWindow, ReduceFn<AccT>> reduceFn) {
        try {
          Set<BoundedWindow> newUnmerged = new HashSet<>(newWindows);
          windowFn.mergeWindows(
              windowFn.new MergeContext() {
                @Override
                public Collection<BoundedWindow> windows() {
                  return Sets.union(buff.keySet(), newWindows);
                }
  
                @Override
                public void merge(Collection<BoundedWindow> merges, BoundedWindow target) {
                  @Nullable
                  MutablePair<Instant, AccT> merged =
                      merges.stream().reduce(null, reduceFn.apply(target), combiner(target));
                  if (merged != null) {
                    buff.put(target, merged);
                  }
                  newUnmerged.removeAll(merges);
                }
              });
          return newUnmerged;
        } catch (Exception e) {
          throw new RuntimeException("Unable to merge accumulators windows", e);
        }
      }
    }
  
    /**
     * Non-merging windowed Spark {@link Aggregator} using a Map of {@link BoundedWindow}s as
     * aggregation buffer. When reducing new input, a windowed accumulator is created for each new
     * window of the input. Otherwise, if the window is known, the accumulators are merged.
     *
     * @param <ValT> {@link CombineFn} input type
     * @param <AccT> {@link CombineFn} accumulator type
     * @param <ResT> {@link CombineFn} / {@link Aggregator} result type
     * @param <InT> {@link Aggregator} input type
     */
    private static class NonMergingWindowedAggregator<ValT, AccT, ResT, InT>
        extends WindowedAggregator<
            ValT, AccT, ResT, InT, BoundedWindow, Map<BoundedWindow, MutablePair<Instant, AccT>>> {
  
      public NonMergingWindowedAggregator(
          CombineFn<ValT, AccT, ResT> combineFn,
          Fun1<WindowedValue<InT>, ValT> valueFn,
          WindowingStrategy<?, ?> windowing,
          Encoder<BoundedWindow> windowEnc,
          Encoder<AccT> accEnc,
          Encoder<WindowedValue<ResT>> outEnc) {
        super(combineFn, valueFn, windowing, windowEnc, accEnc, outEnc, (Class) Map.class);
      }
  
      @Override
      public Map<BoundedWindow, MutablePair<Instant, AccT>> zero() {
        return new HashMap<>();
      }
  
      @Override
      public final Map<BoundedWindow, MutablePair<Instant, AccT>> reduce(
          Map<BoundedWindow, MutablePair<Instant, AccT>> buff, WindowedValue<InT> input) {
        Collection<BoundedWindow> windows = (Collection<BoundedWindow>) input.getWindows();
        return reduce(buff, windows, value(input), input.getTimestamp());
      }
  
      protected Map<BoundedWindow, MutablePair<Instant, AccT>> reduce(
          Map<BoundedWindow, MutablePair<Instant, AccT>> buff,
          Collection<BoundedWindow> windows,
          ValT value,
          Instant timestamp) {
        // for each window add the value to the accumulator
        for (BoundedWindow window : windows) {
          buff.compute(window, (w, acc) -> addToAcc(w, acc, value, timestamp));
        }
        return buff;
      }
  
      @Override
      public Map<BoundedWindow, MutablePair<Instant, AccT>> merge(
          Map<BoundedWindow, MutablePair<Instant, AccT>> b1,
          Map<BoundedWindow, MutablePair<Instant, AccT>> b2) {
        if (b1.isEmpty()) {
          return b2;
        } else if (b2.isEmpty()) {
          return b1;
        }
        if (b2.size() > b1.size()) {
          return merge(b2, b1);
        }
        // merge entries of (smaller) 2nd agg buffer map into first by merging the accumulators
        b2.forEach((w, acc) -> b1.merge(w, acc, combiner(w)));
        return b1;
      }
    }
  
    /**
     * Abstract base of a Spark {@link Aggregator} on {@link WindowedValue}s using a Map of {@link W}
     * as aggregation buffer.
     *
     * @param <ValT> {@link CombineFn} input type
     * @param <AccT> {@link CombineFn} accumulator type
     * @param <ResT> {@link CombineFn} / {@link Aggregator} result type
     * @param <InT> {@link Aggregator} input type
     * @param <W> bounded window type
     * @param <MapT> aggregation buffer {@link W}
     */
    private abstract static class WindowedAggregator<
            ValT,
            AccT,
            ResT,
            InT,
            W extends @NonNull BoundedWindow,
            MapT extends Map<W, @NonNull MutablePair<Instant, AccT>>>
        extends CombineFnAggregator<
            ValT, AccT, ResT, WindowedValue<InT>, MapT, Collection<WindowedValue<ResT>>> {
      private final TimestampCombiner tsCombiner;
  
      public WindowedAggregator(
          CombineFn<ValT, AccT, ResT> combineFn,
          Fun1<WindowedValue<InT>, ValT> valueFn,
          WindowingStrategy<?, ?> windowing,
          Encoder<W> windowEnc,
          Encoder<AccT> accEnc,
          Encoder<WindowedValue<ResT>> outEnc,
          Class<MapT> clazz) {
        super(
            combineFn,
            valueFn,
            mapEncoder(windowEnc, mutablePairEncoder(encoderOf(Instant.class), accEnc), clazz),
            collectionEncoder(outEnc));
        tsCombiner = windowing.getTimestampCombiner();
      }
  
      protected final Instant resolveTimestamp(BoundedWindow w, Instant t1, Instant t2) {
        return tsCombiner.merge(w, t1, t2);
      }
  
      /** Init accumulator with initial input value and timestamp. */
      protected final MutablePair<Instant, AccT> initAcc(ValT value, Instant timestamp) {
        return new MutablePair<>(timestamp, addToAcc(emptyAcc(), value));
      }
  
      /** Merge timestamped accumulators. */
      protected final <T extends MutablePair<Instant, AccT>> @PolyNull T mergeAccs(
          W window, @PolyNull T a1, @PolyNull T a2) {
        if (a1 == null || a2 == null) {
          return a1 == null ? a2 : a1;
        }
        return (T) a1.update(resolveTimestamp(window, a1._1, a2._1), mergeAccs(a1._2, a2._2));
      }
  
      protected BinaryOperator<@Nullable MutablePair<Instant, AccT>> combiner(W target) {
        return (a1, a2) -> mergeAccs(target, a1, a2);
      }
  
      /** Add an input value to a nullable accumulator. */
      protected final MutablePair<Instant, AccT> addToAcc(
          W window, @Nullable MutablePair<Instant, AccT> acc, ValT val, Instant ts) {
        if (acc == null) {
          return initAcc(val, ts);
        }
        return acc.update(resolveTimestamp(window, acc._1, ts), addToAcc(acc._2, val));
      }
  
      @Override
      @SuppressWarnings("nullness") // entries are non null
      public final Collection<WindowedValue<ResT>> finish(MapT buffer) {
        return Collections2.transform(buffer.entrySet(), this::windowedValue);
      }
  
      private WindowedValue<ResT> windowedValue(Entry<W, MutablePair<Instant, AccT>> e) {
        return WindowedValues.of(extract(e.getValue()._2), e.getValue()._1, e.getKey(), NO_FIRING);
      }
    }
  
    /**
     * Abstract base of Spark {@link Aggregator}s using a Beam {@link CombineFn}.
     *
     * @param <ValT> {@link CombineFn} input type
     * @param <AccT> {@link CombineFn} accumulator type
     * @param <ResT> {@link CombineFn} result type
     * @param <InT> {@link Aggregator} input type
     * @param <BuffT> {@link Aggregator} buffer type
     * @param <OutT> {@link Aggregator} output type
     */
    private abstract static class CombineFnAggregator<ValT, AccT, ResT, InT, BuffT, OutT>
        extends Aggregator<InT, BuffT, OutT> {
      private final CombineFn<ValT, AccT, ResT> fn;
      private final Fun1<InT, ValT> valueFn;
      private final Encoder<BuffT> bufferEnc;
      private final Encoder<OutT> outputEnc;
  
      public CombineFnAggregator(
          CombineFn<ValT, AccT, ResT> fn,
          Fun1<InT, ValT> valueFn,
          Encoder<BuffT> bufferEnc,
          Encoder<OutT> outputEnc) {
        this.fn = fn;
        this.valueFn = valueFn;
        this.bufferEnc = bufferEnc;
        this.outputEnc = outputEnc;
      }
  
      protected final ValT value(InT in) {
        return valueFn.apply(in);
      }
  
      protected final AccT emptyAcc() {
        return fn.createAccumulator();
      }
  
      protected final AccT mergeAccs(AccT a1, AccT a2) {
        return fn.mergeAccumulators(ImmutableList.of(a1, a2));
      }
  
      protected final AccT addToAcc(AccT acc, ValT val) {
        return fn.addInput(acc, val);
      }
  
      protected final ResT extract(AccT acc) {
        return fn.extractOutput(acc);
      }
  
      @Override
      public Encoder<BuffT> bufferEncoder() {
        return bufferEnc;
      }
  
      @Override
      public Encoder<OutT> outputEncoder() {
        return outputEnc;
      }
    }
  }
  