/*
 * 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.sdk.values;

import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkState;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.Serializable;
import java.util.AbstractList;
import java.util.AbstractMap;
import java.util.AbstractSet;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.ListIterator;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Objects;
import java.util.PriorityQueue;
import java.util.RandomAccess;
import java.util.Set;
import java.util.SortedMap;
import java.util.function.Supplier;
import org.apache.beam.sdk.annotations.Experimental;
import org.apache.beam.sdk.annotations.Experimental.Kind;
import org.apache.beam.sdk.annotations.Internal;
import org.apache.beam.sdk.coders.BooleanCoder;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.CoderException;
import org.apache.beam.sdk.coders.StructuredCoder;
import org.apache.beam.sdk.io.range.OffsetRange;
import org.apache.beam.sdk.transforms.Materialization;
import org.apache.beam.sdk.transforms.Materializations;
import org.apache.beam.sdk.transforms.Materializations.IterableView;
import org.apache.beam.sdk.transforms.Materializations.MultimapView;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.View;
import org.apache.beam.sdk.transforms.ViewFn;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
import org.apache.beam.sdk.transforms.windowing.WindowMappingFn;
import org.apache.beam.sdk.util.CoderUtils;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.annotations.VisibleForTesting;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.MoreObjects;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Suppliers;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ArrayListMultimap;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.FluentIterable;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableMap;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableSortedMap;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Iterables;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Multimap;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.primitives.Ints;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.primitives.Longs;
import org.checkerframework.checker.nullness.qual.Nullable;

/**
 * <b>For internal use only; no backwards compatibility guarantees.</b>
 *
 * <p>Implementations of {@link PCollectionView} shared across the SDK.
 */
@Internal
@SuppressWarnings({
  "keyfor",
  "nullness", // TODO(https://issues.apache.org/jira/browse/BEAM-10402)
  "rawtypes"
})
public class PCollectionViews {
  public interface TypeDescriptorSupplier<T> extends Supplier<TypeDescriptor<T>>, Serializable {}

  /**
   * Returns a {@code PCollectionView<T>} capable of processing elements windowed using the provided
   * {@link WindowingStrategy}.
   *
   * <p>If {@code hasDefault} is {@code true}, then the view will take on the value {@code
   * defaultValue} for any empty windows.
   */
  public static <T, W extends BoundedWindow> PCollectionView<T> singletonView(
      PCollection<T> pCollection,
      TypeDescriptorSupplier<T> typeDescriptorSupplier,
      WindowingStrategy<?, W> windowingStrategy,
      boolean hasDefault,
      @Nullable T defaultValue,
      Coder<T> defaultValueCoder) {
    return new SimplePCollectionView<>(
        pCollection,
        new SingletonViewFn2<>(hasDefault, defaultValue, defaultValueCoder, typeDescriptorSupplier),
        windowingStrategy.getWindowFn().getDefaultWindowMappingFn(),
        windowingStrategy);
  }

  /**
   * Returns a {@code PCollectionView<T>} capable of processing elements windowed using the provided
   * {@link WindowingStrategy}.
   *
   * <p>If {@code hasDefault} is {@code true}, then the view will take on the value {@code
   * defaultValue} for any empty windows.
   *
   * @deprecated See {@link #singletonView}.
   */
  @Deprecated
  public static <T, W extends BoundedWindow> PCollectionView<T> singletonViewUsingVoidKey(
      PCollection<KV<Void, T>> pCollection,
      TypeDescriptorSupplier<T> typeDescriptorSupplier,
      WindowingStrategy<?, W> windowingStrategy,
      boolean hasDefault,
      @Nullable T defaultValue,
      Coder<T> defaultValueCoder) {
    return new SimplePCollectionView<>(
        pCollection,
        new SingletonViewFn<>(hasDefault, defaultValue, defaultValueCoder, typeDescriptorSupplier),
        windowingStrategy.getWindowFn().getDefaultWindowMappingFn(),
        windowingStrategy);
  }

  /**
   * Returns a {@code PCollectionView<Iterable<T>>} capable of processing elements windowed using
   * the provided {@link WindowingStrategy}.
   */
  public static <T, W extends BoundedWindow> PCollectionView<Iterable<T>> iterableView(
      PCollection<T> pCollection,
      TypeDescriptorSupplier<T> typeDescriptorSupplier,
      WindowingStrategy<?, W> windowingStrategy) {
    return new SimplePCollectionView<>(
        pCollection,
        new IterableViewFn2<>(typeDescriptorSupplier),
        windowingStrategy.getWindowFn().getDefaultWindowMappingFn(),
        windowingStrategy);
  }

  /**
   * Returns a {@code PCollectionView<Iterable<T>>} capable of processing elements windowed using
   * the provided {@link WindowingStrategy}.
   *
   * @deprecated See {@link #iterableView}.
   */
  @Deprecated
  public static <T, W extends BoundedWindow> PCollectionView<Iterable<T>> iterableViewUsingVoidKey(
      PCollection<KV<Void, T>> pCollection,
      TypeDescriptorSupplier<T> typeDescriptorSupplier,
      WindowingStrategy<?, W> windowingStrategy) {
    return new SimplePCollectionView<>(
        pCollection,
        new IterableViewFn<>(typeDescriptorSupplier),
        windowingStrategy.getWindowFn().getDefaultWindowMappingFn(),
        windowingStrategy);
  }

  /**
   * Returns a {@code PCollectionView<List<T>>} capable of processing elements windowed using the
   * provided {@link WindowingStrategy}.
   */
  public static <T, W extends BoundedWindow> PCollectionView<List<T>> listView(
      PCollection<KV<Long, ValueOrMetadata<T, OffsetRange>>> pCollection,
      TypeDescriptorSupplier<T> typeDescriptorSupplier,
      WindowingStrategy<?, W> windowingStrategy) {
    return new SimplePCollectionView<>(
        pCollection,
        new ListViewFn2<>(typeDescriptorSupplier),
        windowingStrategy.getWindowFn().getDefaultWindowMappingFn(),
        windowingStrategy);
  }

  /**
   * Returns a {@code PCollectionView<List<T>>} capable of processing elements windowed using the
   * provided {@link WindowingStrategy}.
   *
   * @deprecated See {@link #listView}.
   */
  @Deprecated
  public static <T, W extends BoundedWindow> PCollectionView<List<T>> listViewUsingVoidKey(
      PCollection<KV<Void, T>> pCollection,
      TypeDescriptorSupplier<T> typeDescriptorSupplier,
      WindowingStrategy<?, W> windowingStrategy) {
    return new SimplePCollectionView<>(
        pCollection,
        new ListViewFn<>(typeDescriptorSupplier),
        windowingStrategy.getWindowFn().getDefaultWindowMappingFn(),
        windowingStrategy);
  }

  /**
   * Returns a {@code PCollectionView<List<T>>} capable of processing elements windowed using the
   * provided {@link WindowingStrategy}.
   *
   * @deprecated See {@link #listView}.
   */
  @Deprecated
  public static <T, W extends BoundedWindow> PCollectionView<List<T>> listViewUsingVoidKey(
      PCollection<KV<Void, T>> pCollection,
      TupleTag<MultimapView<Void, T>> tag,
      TypeDescriptorSupplier<T> typeDescriptorSupplier,
      WindowingStrategy<?, W> windowingStrategy) {
    return new SimplePCollectionView<>(
        pCollection,
        tag,
        new ListViewFn<>(typeDescriptorSupplier),
        windowingStrategy.getWindowFn().getDefaultWindowMappingFn(),
        windowingStrategy);
  }

  /**
   * Returns a {@code PCollectionView<Map<K, V>>} capable of processing elements windowed using the
   * provided {@link WindowingStrategy}.
   */
  public static <K, V, W extends BoundedWindow> PCollectionView<Map<K, V>> mapView(
      PCollection<KV<K, V>> pCollection,
      TypeDescriptorSupplier<K> keyTypeDescriptorSupplier,
      TypeDescriptorSupplier<V> valueTypeDescriptorSupplier,
      WindowingStrategy<?, W> windowingStrategy) {
    return new SimplePCollectionView<>(
        pCollection,
        new MapViewFn2<>(keyTypeDescriptorSupplier, valueTypeDescriptorSupplier),
        windowingStrategy.getWindowFn().getDefaultWindowMappingFn(),
        windowingStrategy);
  }

  /**
   * Returns a {@code PCollectionView<Map<K, V>>} capable of processing elements windowed using the
   * provided {@link WindowingStrategy}.
   *
   * @deprecated See {@link #mapView}.
   */
  @Deprecated
  public static <K, V, W extends BoundedWindow> PCollectionView<Map<K, V>> mapViewUsingVoidKey(
      PCollection<KV<Void, KV<K, V>>> pCollection,
      TypeDescriptorSupplier<K> keyTypeDescriptorSupplier,
      TypeDescriptorSupplier<V> valueTypeDescriptorSupplier,
      WindowingStrategy<?, W> windowingStrategy) {
    return new SimplePCollectionView<>(
        pCollection,
        new MapViewFn<>(keyTypeDescriptorSupplier, valueTypeDescriptorSupplier),
        windowingStrategy.getWindowFn().getDefaultWindowMappingFn(),
        windowingStrategy);
  }

  /**
   * Returns a {@code PCollectionView<Map<K, Iterable<V>>>} capable of processing elements windowed
   * using the provided {@link WindowingStrategy}.
   */
  public static <K, V, W extends BoundedWindow> PCollectionView<Map<K, Iterable<V>>> multimapView(
      PCollection<KV<K, V>> pCollection,
      TypeDescriptorSupplier<K> keyTypeDescriptorSupplier,
      TypeDescriptorSupplier<V> valueTypeDescriptorSupplier,
      WindowingStrategy<?, W> windowingStrategy) {
    return new SimplePCollectionView<>(
        pCollection,
        new MultimapViewFn2<>(keyTypeDescriptorSupplier, valueTypeDescriptorSupplier),
        windowingStrategy.getWindowFn().getDefaultWindowMappingFn(),
        windowingStrategy);
  }

  /**
   * Returns a {@code PCollectionView<Map<K, Iterable<V>>>} capable of processing elements windowed
   * using the provided {@link WindowingStrategy}.
   *
   * @deprecated See {@link #multimapView}.
   */
  @Deprecated
  public static <K, V, W extends BoundedWindow>
      PCollectionView<Map<K, Iterable<V>>> multimapViewUsingVoidKey(
          PCollection<KV<Void, KV<K, V>>> pCollection,
          TypeDescriptorSupplier<K> keyTypeDescriptorSupplier,
          TypeDescriptorSupplier<V> valueTypeDescriptorSupplier,
          WindowingStrategy<?, W> windowingStrategy) {
    return new SimplePCollectionView<>(
        pCollection,
        new MultimapViewFn<>(keyTypeDescriptorSupplier, valueTypeDescriptorSupplier),
        windowingStrategy.getWindowFn().getDefaultWindowMappingFn(),
        windowingStrategy);
  }

  /**
   * Expands a list of {@link PCollectionView} into the form needed for {@link
   * PTransform#getAdditionalInputs()}.
   */
  public static Map<TupleTag<?>, PValue> toAdditionalInputs(Iterable<PCollectionView<?>> views) {
    ImmutableMap.Builder<TupleTag<?>, PValue> additionalInputs = ImmutableMap.builder();
    for (PCollectionView<?> view : views) {
      additionalInputs.put(view.getTagInternal(), view.getPCollection());
    }
    return additionalInputs.build();
  }

  /**
   * Implementation which is able to adapt an iterable materialization to a {@code T}.
   *
   * <p>For internal use only.
   *
   * <p>Instantiate via {@link PCollectionViews#singletonView}.
   *
   * <p>{@link SingletonViewFn} is meant to be removed in the future and replaced with this class.
   */
  @Experimental(Kind.CORE_RUNNERS_ONLY)
  private static class SingletonViewFn2<T> extends ViewFn<IterableView<T>, T> {
    private byte @Nullable [] encodedDefaultValue;
    private transient @Nullable T defaultValue;
    private @Nullable Coder<T> valueCoder;
    private boolean hasDefault;
    private TypeDescriptorSupplier<T> typeDescriptorSupplier;

    private SingletonViewFn2(
        boolean hasDefault,
        T defaultValue,
        Coder<T> valueCoder,
        TypeDescriptorSupplier<T> typeDescriptorSupplier) {
      this.hasDefault = hasDefault;
      this.defaultValue = defaultValue;
      this.valueCoder = valueCoder;
      this.typeDescriptorSupplier = typeDescriptorSupplier;
      if (hasDefault) {
        try {
          this.encodedDefaultValue = CoderUtils.encodeToByteArray(valueCoder, defaultValue);
        } catch (IOException e) {
          throw new RuntimeException("Unexpected IOException: ", e);
        }
      }
    }

    /** Returns if a default value was specified. */
    @Internal
    public boolean hasDefault() {
      return hasDefault;
    }

    /**
     * Returns the default value that was specified.
     *
     * <p>For internal use only.
     *
     * @throws NoSuchElementException if no default was specified.
     */
    public T getDefaultValue() {
      if (!hasDefault) {
        throw new NoSuchElementException("Empty PCollection accessed as a singleton view.");
      }
      // Lazily decode the default value once
      synchronized (this) {
        if (encodedDefaultValue != null) {
          try {
            defaultValue = CoderUtils.decodeFromByteArray(valueCoder, encodedDefaultValue);
            // Clear the encoded default value to free the reference once we have the object
            // version. Also, this will guarantee that the value will only be decoded once.
            encodedDefaultValue = null;
          } catch (IOException e) {
            throw new RuntimeException("Unexpected IOException: ", e);
          }
        }
        return defaultValue;
      }
    }

    @Override
    public Materialization<IterableView<T>> getMaterialization() {
      return Materializations.iterable();
    }

    @Override
    public T apply(IterableView<T> primitiveViewT) {
      try {
        return Iterables.getOnlyElement(primitiveViewT.get());
      } catch (NoSuchElementException exc) {
        return getDefaultValue();
      } catch (IllegalArgumentException exc) {
        throw new IllegalArgumentException(
            "PCollection with more than one element accessed as a singleton view.");
      }
    }

    @Override
    public TypeDescriptor<T> getTypeDescriptor() {
      return typeDescriptorSupplier.get();
    }
  }

  /**
   * Implementation which is able to adapt a multimap materialization to a {@code T}.
   *
   * <p>For internal use only.
   *
   * @deprecated See {@link SingletonViewFn2}.
   */
  @Deprecated
  @Experimental(Kind.CORE_RUNNERS_ONLY)
  public static class SingletonViewFn<T> extends ViewFn<MultimapView<Void, T>, T> {
    private byte @Nullable [] encodedDefaultValue;
    private transient @Nullable T defaultValue;
    private @Nullable Coder<T> valueCoder;
    private boolean hasDefault;
    private TypeDescriptorSupplier<T> typeDescriptorSupplier;

    private SingletonViewFn(
        boolean hasDefault,
        T defaultValue,
        Coder<T> valueCoder,
        TypeDescriptorSupplier<T> typeDescriptorSupplier) {
      this.hasDefault = hasDefault;
      this.defaultValue = defaultValue;
      this.valueCoder = valueCoder;
      this.typeDescriptorSupplier = typeDescriptorSupplier;
      if (hasDefault) {
        try {
          this.encodedDefaultValue = CoderUtils.encodeToByteArray(valueCoder, defaultValue);
        } catch (IOException e) {
          throw new RuntimeException("Unexpected IOException: ", e);
        }
      }
    }

    /** Returns if a default value was specified. */
    @Internal
    public boolean hasDefault() {
      return hasDefault;
    }

    /**
     * Returns the default value that was specified.
     *
     * <p>For internal use only.
     *
     * @throws NoSuchElementException if no default was specified.
     */
    public T getDefaultValue() {
      if (!hasDefault) {
        throw new NoSuchElementException("Empty PCollection accessed as a singleton view.");
      }
      // Lazily decode the default value once
      synchronized (this) {
        if (encodedDefaultValue != null) {
          try {
            defaultValue = CoderUtils.decodeFromByteArray(valueCoder, encodedDefaultValue);
            // Clear the encoded default value to free the reference once we have the object
            // version. Also, this will guarantee that the value will only be decoded once.
            encodedDefaultValue = null;
          } catch (IOException e) {
            throw new RuntimeException("Unexpected IOException: ", e);
          }
        }
        return defaultValue;
      }
    }

    @Override
    public Materialization<MultimapView<Void, T>> getMaterialization() {
      return Materializations.multimap();
    }

    @Override
    public T apply(MultimapView<Void, T> primitiveViewT) {
      try {
        return Iterables.getOnlyElement(primitiveViewT.get(null));
      } catch (NoSuchElementException exc) {
        return getDefaultValue();
      } catch (IllegalArgumentException exc) {
        throw new IllegalArgumentException(
            "PCollection with more than one element accessed as a singleton view.");
      }
    }

    @Override
    public TypeDescriptor<T> getTypeDescriptor() {
      return typeDescriptorSupplier.get();
    }
  }

  /**
   * Implementation which is able to adapt an iterable materialization to a {@code Iterable<T>}.
   *
   * <p>For internal use only.
   *
   * <p>Instantiate via {@link PCollectionViews#iterableView}.
   *
   * <p>{@link IterableViewFn} is meant to be removed in the future and replaced with this class.
   */
  @Experimental(Kind.CORE_RUNNERS_ONLY)
  private static class IterableViewFn2<T> extends ViewFn<IterableView<T>, Iterable<T>> {
    private TypeDescriptorSupplier<T> typeDescriptorSupplier;

    public IterableViewFn2(TypeDescriptorSupplier<T> typeDescriptorSupplier) {
      this.typeDescriptorSupplier = typeDescriptorSupplier;
    }

    @Override
    public Materialization<IterableView<T>> getMaterialization() {
      return Materializations.iterable();
    }

    @Override
    public Iterable<T> apply(IterableView<T> primitiveViewT) {
      return Iterables.unmodifiableIterable(primitiveViewT.get());
    }

    @Override
    public TypeDescriptor<Iterable<T>> getTypeDescriptor() {
      return TypeDescriptors.iterables(typeDescriptorSupplier.get());
    }
  }

  /**
   * Implementation which is able to adapt a multimap materialization to a {@code Iterable<T>}.
   *
   * <p>For internal use only.
   *
   * @deprecated See {@link IterableViewFn2}.
   */
  @Deprecated
  @Experimental(Kind.CORE_RUNNERS_ONLY)
  public static class IterableViewFn<T> extends ViewFn<MultimapView<Void, T>, Iterable<T>> {
    private TypeDescriptorSupplier<T> typeDescriptorSupplier;

    public IterableViewFn(TypeDescriptorSupplier<T> typeDescriptorSupplier) {
      this.typeDescriptorSupplier = typeDescriptorSupplier;
    }

    @Override
    public Materialization<MultimapView<Void, T>> getMaterialization() {
      return Materializations.multimap();
    }

    @Override
    public Iterable<T> apply(MultimapView<Void, T> primitiveViewT) {
      return Iterables.unmodifiableIterable(primitiveViewT.get(null));
    }

    @Override
    public TypeDescriptor<Iterable<T>> getTypeDescriptor() {
      return TypeDescriptors.iterables(typeDescriptorSupplier.get());
    }
  }

  /**
   * Implementation which is able to adapt a multimap materialization to a {@code List<T>}.
   *
   * <p>For internal use only.
   *
   * <p>Instantiate via {@link PCollectionViews#listView}.
   *
   * <p>{@link ListViewFn} is meant to be removed in the future and replaced with this class.
   */
  @Experimental(Kind.CORE_RUNNERS_ONLY)
  @VisibleForTesting
  static class ListViewFn2<T>
      extends ViewFn<MultimapView<Long, ValueOrMetadata<T, OffsetRange>>, List<T>> {
    private TypeDescriptorSupplier<T> typeDescriptorSupplier;

    public ListViewFn2(TypeDescriptorSupplier<T> typeDescriptorSupplier) {
      this.typeDescriptorSupplier = typeDescriptorSupplier;
    }

    @Override
    public Materialization<MultimapView<Long, ValueOrMetadata<T, OffsetRange>>>
        getMaterialization() {
      return Materializations.multimap();
    }

    @Override
    public List<T> apply(MultimapView<Long, ValueOrMetadata<T, OffsetRange>> primitiveViewT) {
      return Collections.unmodifiableList(new ListOverMultimapView<>(primitiveViewT));
    }

    @Override
    public TypeDescriptor<List<T>> getTypeDescriptor() {
      return TypeDescriptors.lists(typeDescriptorSupplier.get());
    }

    /**
     * A {@link List} adapter over a {@link MultimapView}.
     *
     * <p>See {@link View.AsList} for a description of the materialized format and {@code index} to
     * {@code (position, sub-position)} mapping details.
     */
    private static class ListOverMultimapView<T> extends AbstractList<T> implements RandomAccess {
      private final MultimapView<Long, ValueOrMetadata<T, OffsetRange>> primitiveView;
      /**
       * A mapping from non over-lapping ranges to the number of elements at each position within
       * that range. Ranges not specified in the mapping implicitly have 0 elements at those
       * positions.
       *
       * <p>Used to quickly compute the {@code index} -> {@code (position, sub-position} within the
       * map.
       */
      private final Supplier<SortedMap<OffsetRange, Integer>>
          nonOverlappingRangesToNumElementsPerPosition;

      private final Supplier<Integer> size;

      private ListOverMultimapView(
          MultimapView<Long, ValueOrMetadata<T, OffsetRange>> primitiveView) {
        this.primitiveView = primitiveView;
        this.nonOverlappingRangesToNumElementsPerPosition =
            Suppliers.memoize(
                () ->
                    computeOverlappingRanges(
                        Iterables.transform(
                            primitiveView.get(Long.MIN_VALUE), (value) -> value.getMetadata())));
        this.size =
            Suppliers.memoize(
                () -> computeTotalNumElements(nonOverlappingRangesToNumElementsPerPosition.get()));
      }

      @Override
      public T get(int index) {
        if (index < 0 || index >= size.get()) {
          throw new IndexOutOfBoundsException();
        }
        KV<Long, Integer> position =
            computePositionForIndex(nonOverlappingRangesToNumElementsPerPosition.get(), index);
        return Iterables.get(primitiveView.get(position.getKey()), position.getValue()).get();
      }

      @Override
      public int size() {
        return size.get();
      }

      @Override
      public Iterator<T> iterator() {
        return listIterator();
      }

      @Override
      public ListIterator<T> listIterator() {
        return super.listIterator();
      }

      /** A {@link ListIterator} over {@link MultimapView} adapter. */
      private class ListIteratorOverMultimapView implements ListIterator<T> {
        private int position;

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

        @Override
        public T next() {
          if (!hasNext()) {
            throw new NoSuchElementException();
          }
          T rval = get(position);
          position += 1;
          return rval;
        }

        @Override
        public boolean hasPrevious() {
          return position > 0;
        }

        @Override
        public T previous() {
          if (!hasPrevious()) {
            throw new NoSuchElementException();
          }
          position -= 1;
          return get(position);
        }

        @Override
        public int nextIndex() {
          return position;
        }

        @Override
        public int previousIndex() {
          return position - 1;
        }

        @Override
        public void remove() {
          throw new UnsupportedOperationException();
        }

        @Override
        public void set(T e) {
          throw new UnsupportedOperationException();
        }

        @Override
        public void add(T e) {
          throw new UnsupportedOperationException();
        }
      }
    }
  }

  /**
   * Compares {@link OffsetRange}s such that ranges are ordered by the smallest {@code from} and in
   * case of a tie the smallest {@code to}.
   */
  @VisibleForTesting
  static class OffsetRangeComparator implements Comparator<OffsetRange> {
    private static final OffsetRangeComparator INSTANCE = new OffsetRangeComparator();

    @Override
    public int compare(OffsetRange o1, OffsetRange o2) {
      int fromComparison = Longs.compare(o1.getFrom(), o2.getFrom());
      if (fromComparison != 0) {
        return fromComparison;
      }
      return Longs.compare(o1.getTo(), o2.getTo());
    }
  }

  @VisibleForTesting
  static SortedMap<OffsetRange, Integer> computeOverlappingRanges(Iterable<OffsetRange> ranges) {
    ImmutableSortedMap.Builder<OffsetRange, Integer> rval =
        ImmutableSortedMap.orderedBy(OffsetRangeComparator.INSTANCE);
    List<OffsetRange> sortedRanges = Lists.newArrayList(ranges);
    if (sortedRanges.isEmpty()) {
      return rval.build();
    }
    Collections.sort(sortedRanges, OffsetRangeComparator.INSTANCE);

    // Stores ranges in smallest 'from' and then smallest 'to' order
    // e.g. [2, 7), [3, 4), [3, 5), [3, 5), [3, 6), [4, 0)
    PriorityQueue<OffsetRange> rangesWithSameFrom =
        new PriorityQueue<>(OffsetRangeComparator.INSTANCE);
    Iterator<OffsetRange> iterator = sortedRanges.iterator();

    // Stored in reverse sorted order so that when we iterate and re-add them back to
    // overlappingRanges they are stored in sorted order from smallest to largest range.to
    List<OffsetRange> rangesToProcess = new ArrayList<>();
    while (iterator.hasNext()) {
      OffsetRange current = iterator.next();
      // Skip empty ranges
      if (current.getFrom() == current.getTo()) {
        continue;
      }

      // If the current range has a different 'from' then a prior range then we must produce
      // ranges in [rangesWithSameFrom.from, current.from)
      while (!rangesWithSameFrom.isEmpty()
          && rangesWithSameFrom.peek().getFrom() != current.getFrom()) {
        rangesToProcess.addAll(rangesWithSameFrom);
        Collections.sort(rangesToProcess, OffsetRangeComparator.INSTANCE);
        rangesWithSameFrom.clear();

        int i = 0;
        long lastTo = rangesToProcess.get(i).getFrom();
        // Output all the ranges that are strictly less then current.from
        // e.g. current.to := 7 for [3, 4), [3, 5), [3, 5), [3, 6) will produce
        // [3, 4) := 4
        // [4, 5) := 3
        // [5, 6) := 1
        for (; i < rangesToProcess.size(); ++i) {
          if (rangesToProcess.get(i).getTo() > current.getFrom()) {
            break;
          }
          // Output only the first of any subsequent duplicate ranges
          if (i == 0 || rangesToProcess.get(i - 1).getTo() != rangesToProcess.get(i).getTo()) {
            rval.put(
                new OffsetRange(lastTo, rangesToProcess.get(i).getTo()),
                rangesToProcess.size() - i);
            lastTo = rangesToProcess.get(i).getTo();
          }
        }

        // We exitted the loop with 'to' > current.from, we must add the range [lastTo,
        // current.from) if it is non-empty
        if (lastTo < current.getFrom() && i != rangesToProcess.size()) {
          rval.put(new OffsetRange(lastTo, current.getFrom()), rangesToProcess.size() - i);
        }

        // The remaining ranges have a 'to' that is greater then 'current.from' and will overlap
        // with current so add them back to rangesWithSameFrom with the updated 'from'
        for (; i < rangesToProcess.size(); ++i) {
          rangesWithSameFrom.add(
              new OffsetRange(current.getFrom(), rangesToProcess.get(i).getTo()));
        }

        rangesToProcess.clear();
      }
      rangesWithSameFrom.add(current);
    }

    // Process the last chunk of overlapping ranges
    while (!rangesWithSameFrom.isEmpty()) {
      // This range always represents the range with with the smallest 'to'
      OffsetRange current = rangesWithSameFrom.remove();

      rangesToProcess.addAll(rangesWithSameFrom);
      Collections.sort(rangesToProcess, OffsetRangeComparator.INSTANCE);
      rangesWithSameFrom.clear();

      rval.put(current, rangesToProcess.size() + 1 /* include current */);

      // Shorten all the remaining ranges such that they start with current.to
      for (OffsetRange rangeWithDifferentFrom : rangesToProcess) {
        // Skip any duplicates of current
        if (rangeWithDifferentFrom.getTo() > current.getTo()) {
          rangesWithSameFrom.add(new OffsetRange(current.getTo(), rangeWithDifferentFrom.getTo()));
        }
      }
      rangesToProcess.clear();
    }
    return rval.build();
  }

  @VisibleForTesting
  static int computeTotalNumElements(
      Map<OffsetRange, Integer> nonOverlappingRangesToNumElementsPerPosition) {
    long sum = 0;
    for (Map.Entry<OffsetRange, Integer> range :
        nonOverlappingRangesToNumElementsPerPosition.entrySet()) {
      sum +=
          Math.multiplyExact(
              Math.subtractExact(range.getKey().getTo(), range.getKey().getFrom()),
              range.getValue());
    }
    return Ints.checkedCast(sum);
  }

  @VisibleForTesting
  static KV<Long, Integer> computePositionForIndex(
      Map<OffsetRange, Integer> nonOverlappingRangesToNumElementsPerPosition, int index) {
    if (index < 0) {
      throw new IndexOutOfBoundsException(
          String.format(
              "Position %s was out of bounds for ranges %s.",
              index, nonOverlappingRangesToNumElementsPerPosition));
    }
    for (Map.Entry<OffsetRange, Integer> range :
        nonOverlappingRangesToNumElementsPerPosition.entrySet()) {
      int numElementsInRange =
          Ints.checkedCast(
              Math.multiplyExact(
                  Math.subtractExact(range.getKey().getTo(), range.getKey().getFrom()),
                  range.getValue()));
      if (numElementsInRange <= index) {
        index -= numElementsInRange;
        continue;
      }
      long position = range.getKey().getFrom() + index / range.getValue();
      int subPosition = index % range.getValue();
      return KV.of(position, subPosition);
    }
    throw new IndexOutOfBoundsException(
        String.format(
            "Position %s was out of bounds for ranges %s.",
            index, nonOverlappingRangesToNumElementsPerPosition));
  }

  /** Stores values or metadata about values. */
  public static class ValueOrMetadata<T, MetaT> {
    private final T value;
    private final boolean isMetadata;
    private final MetaT metadata;

    public static <T, MetaT> ValueOrMetadata<T, MetaT> create(T value) {
      return new ValueOrMetadata(false, value, null);
    }

    public static <T, MetaT> ValueOrMetadata<T, MetaT> createMetadata(MetaT metadata) {
      return new ValueOrMetadata(true, null, metadata);
    }

    public ValueOrMetadata(boolean isMetadata, T value, MetaT metadata) {
      this.isMetadata = isMetadata;
      this.value = value;
      this.metadata = metadata;
    }

    public T get() {
      checkState(!isMetadata);
      return value;
    }

    public boolean isMetadata() {
      return isMetadata;
    }

    public MetaT getMetadata() {
      checkState(isMetadata);
      return metadata;
    }
  }

  /** A coder for {@link ValueOrMetadata}. */
  public static class ValueOrMetadataCoder<T, MetaT>
      extends StructuredCoder<ValueOrMetadata<T, MetaT>> {
    public static <T, MetaT> ValueOrMetadataCoder<T, MetaT> create(
        Coder<T> valueCoder, Coder<MetaT> metadataCoder) {
      return new ValueOrMetadataCoder<>(valueCoder, metadataCoder);
    }

    private final Coder<T> valueCoder;
    private final Coder<MetaT> metadataCoder;

    private ValueOrMetadataCoder(Coder<T> valueCoder, Coder<MetaT> metadataCoder) {
      this.valueCoder = valueCoder;
      this.metadataCoder = metadataCoder;
    }

    @Override
    public void encode(ValueOrMetadata<T, MetaT> value, OutputStream outStream)
        throws CoderException, IOException {
      BooleanCoder.of().encode(value.isMetadata(), outStream);
      if (value.isMetadata()) {
        metadataCoder.encode(value.getMetadata(), outStream);
      } else {
        valueCoder.encode(value.get(), outStream);
      }
    }

    @Override
    public ValueOrMetadata<T, MetaT> decode(InputStream inStream)
        throws CoderException, IOException {
      boolean isMetadata = BooleanCoder.of().decode(inStream);
      if (isMetadata) {
        return ValueOrMetadata.createMetadata(metadataCoder.decode(inStream));
      } else {
        return ValueOrMetadata.create(valueCoder.decode(inStream));
      }
    }

    @Override
    public List<? extends Coder<?>> getCoderArguments() {
      return Arrays.asList(valueCoder, metadataCoder);
    }

    @Override
    public void verifyDeterministic() throws NonDeterministicException {
      verifyDeterministic(valueCoder, "value coder");
      verifyDeterministic(metadataCoder, "metadata coder");
    }
  }

  /**
   * Implementation which is able to adapt a multimap materialization to a {@code List<T>}.
   *
   * <p>For internal use only.
   *
   * @deprecated See {@link ListViewFn2}.
   */
  @Deprecated
  @Experimental(Kind.CORE_RUNNERS_ONLY)
  public static class ListViewFn<T> extends ViewFn<MultimapView<Void, T>, List<T>> {
    private TypeDescriptorSupplier<T> typeDescriptorSupplier;

    public ListViewFn(TypeDescriptorSupplier<T> typeDescriptorSupplier) {
      this.typeDescriptorSupplier = typeDescriptorSupplier;
    }

    @Override
    public Materialization<MultimapView<Void, T>> getMaterialization() {
      return Materializations.multimap();
    }

    @Override
    public List<T> apply(MultimapView<Void, T> primitiveViewT) {
      List<T> list = new ArrayList<>();
      for (T t : primitiveViewT.get(null)) {
        list.add(t);
      }
      return Collections.unmodifiableList(list);
    }

    @Override
    public TypeDescriptor<List<T>> getTypeDescriptor() {
      return TypeDescriptors.lists(typeDescriptorSupplier.get());
    }

    @Override
    public boolean equals(@Nullable Object other) {
      return other instanceof ListViewFn;
    }

    @Override
    public int hashCode() {
      return ListViewFn.class.hashCode();
    }
  }

  /**
   * Implementation which is able to adapt a multimap materialization to a {@code Map<K,
   * Iterable<V>>}.
   *
   * <p>For internal use only.
   *
   * <p>Instantiate via {@link PCollectionViews#multimapView}.
   *
   * <p>{@link MultimapViewFn} is meant to be removed in the future and replaced with this class.
   */
  @Experimental(Kind.CORE_RUNNERS_ONLY)
  private static class MultimapViewFn2<K, V>
      extends ViewFn<MultimapView<K, V>, Map<K, Iterable<V>>> {
    private TypeDescriptorSupplier<K> keyTypeDescriptorSupplier;
    private TypeDescriptorSupplier<V> valueTypeDescriptorSupplier;

    public MultimapViewFn2(
        TypeDescriptorSupplier<K> keyTypeDescriptorSupplier,
        TypeDescriptorSupplier<V> valueTypeDescriptorSupplier) {
      this.keyTypeDescriptorSupplier = keyTypeDescriptorSupplier;
      this.valueTypeDescriptorSupplier = valueTypeDescriptorSupplier;
    }

    @Override
    public Materialization<MultimapView<K, V>> getMaterialization() {
      return Materializations.multimap();
    }

    @Override
    public Map<K, Iterable<V>> apply(MultimapView<K, V> primitiveViewT) {
      return Collections.unmodifiableMap(new MultimapViewToMultimapAdapter<>(primitiveViewT));
    }

    @Override
    public TypeDescriptor<Map<K, Iterable<V>>> getTypeDescriptor() {
      return TypeDescriptors.maps(
          keyTypeDescriptorSupplier.get(),
          TypeDescriptors.iterables(valueTypeDescriptorSupplier.get()));
    }
  }

  /**
   * Implementation which is able to adapt a multimap materialization to a {@code Map<K,
   * Iterable<V>>}.
   *
   * <p>For internal use only.
   *
   * @deprecated See {@link MultimapViewFn2}.
   */
  @Deprecated
  @Experimental(Kind.CORE_RUNNERS_ONLY)
  public static class MultimapViewFn<K, V>
      extends ViewFn<MultimapView<Void, KV<K, V>>, Map<K, Iterable<V>>> {
    private TypeDescriptorSupplier<K> keyTypeDescriptorSupplier;
    private TypeDescriptorSupplier<V> valueTypeDescriptorSupplier;

    public MultimapViewFn(
        TypeDescriptorSupplier<K> keyTypeDescriptorSupplier,
        TypeDescriptorSupplier<V> valueTypeDescriptorSupplier) {
      this.keyTypeDescriptorSupplier = keyTypeDescriptorSupplier;
      this.valueTypeDescriptorSupplier = valueTypeDescriptorSupplier;
    }

    @Override
    public Materialization<MultimapView<Void, KV<K, V>>> getMaterialization() {
      return Materializations.multimap();
    }

    @Override
    public Map<K, Iterable<V>> apply(MultimapView<Void, KV<K, V>> primitiveViewT) {
      // TODO: BEAM-3071 - fix this so that we aren't relying on Java equality and are
      // using structural value equality.
      Multimap<K, V> multimap = ArrayListMultimap.create();
      for (KV<K, V> elem : primitiveViewT.get(null)) {
        multimap.put(elem.getKey(), elem.getValue());
      }
      // Safe covariant cast that Java cannot express without rawtypes, even with unchecked casts
      @SuppressWarnings({"unchecked", "rawtypes"})
      Map<K, Iterable<V>> resultMap = (Map) multimap.asMap();
      return Collections.unmodifiableMap(resultMap);
    }

    @Override
    public TypeDescriptor<Map<K, Iterable<V>>> getTypeDescriptor() {
      return TypeDescriptors.maps(
          keyTypeDescriptorSupplier.get(),
          TypeDescriptors.iterables(valueTypeDescriptorSupplier.get()));
    }
  }

  /**
   * Implementation which is able to adapt a multimap materialization to a {@code Map<K, V>}.
   *
   * <p>For internal use only.
   *
   * <p>Instantiate via {@link PCollectionViews#mapView}.
   *
   * <p>{@link MapViewFn} is meant to be removed in the future and replaced with this class.
   */
  private static class MapViewFn2<K, V> extends ViewFn<MultimapView<K, V>, Map<K, V>> {
    private TypeDescriptorSupplier<K> keyTypeDescriptorSupplier;
    private TypeDescriptorSupplier<V> valueTypeDescriptorSupplier;

    public MapViewFn2(
        TypeDescriptorSupplier<K> keyTypeDescriptorSupplier,
        TypeDescriptorSupplier<V> valueTypeDescriptorSupplier) {
      this.keyTypeDescriptorSupplier = keyTypeDescriptorSupplier;
      this.valueTypeDescriptorSupplier = valueTypeDescriptorSupplier;
    }

    @Override
    public Materialization<MultimapView<K, V>> getMaterialization() {
      return Materializations.multimap();
    }

    @Override
    public Map<K, V> apply(MultimapView<K, V> primitiveViewT) {
      return Collections.unmodifiableMap(new MultimapViewToMapAdapter<>(primitiveViewT));
    }

    @Override
    public TypeDescriptor<Map<K, V>> getTypeDescriptor() {
      return TypeDescriptors.maps(
          keyTypeDescriptorSupplier.get(), valueTypeDescriptorSupplier.get());
    }
  }

  /**
   * Implementation which is able to adapt a multimap materialization to a {@code Map<K, V>}.
   *
   * <p>For internal use only.
   *
   * @deprecated See {@link MapViewFn2}.
   */
  @Deprecated
  @Experimental(Kind.CORE_RUNNERS_ONLY)
  public static class MapViewFn<K, V> extends ViewFn<MultimapView<Void, KV<K, V>>, Map<K, V>> {
    private TypeDescriptorSupplier<K> keyTypeDescriptorSupplier;
    private TypeDescriptorSupplier<V> valueTypeDescriptorSupplier;

    public MapViewFn(
        TypeDescriptorSupplier<K> keyTypeDescriptorSupplier,
        TypeDescriptorSupplier<V> valueTypeDescriptorSupplier) {
      this.keyTypeDescriptorSupplier = keyTypeDescriptorSupplier;
      this.valueTypeDescriptorSupplier = valueTypeDescriptorSupplier;
    }

    @Override
    public Materialization<MultimapView<Void, KV<K, V>>> getMaterialization() {
      return Materializations.multimap();
    }

    @Override
    public Map<K, V> apply(MultimapView<Void, KV<K, V>> primitiveViewT) {
      // TODO: BEAM-3071 - fix this so that we aren't relying on Java equality and are
      // using structural value equality.
      Map<K, V> map = new HashMap<>();
      for (KV<K, V> elem : primitiveViewT.get(null)) {
        if (map.containsKey(elem.getKey())) {
          throw new IllegalArgumentException("Duplicate values for " + elem.getKey());
        }
        map.put(elem.getKey(), elem.getValue());
      }
      return Collections.unmodifiableMap(map);
    }

    @Override
    public TypeDescriptor<Map<K, V>> getTypeDescriptor() {
      return TypeDescriptors.maps(
          keyTypeDescriptorSupplier.get(), valueTypeDescriptorSupplier.get());
    }
  }

  /**
   * A class for {@link PCollectionView} implementations, with additional type parameters that are
   * not visible at pipeline assembly time when the view is used as a side input.
   *
   * <p>For internal use only.
   */
  public static class SimplePCollectionView<ElemT, PrimitiveViewT, ViewT, W extends BoundedWindow>
      extends PValueBase implements PCollectionView<ViewT> {
    /** The {@link PCollection} this view was originally created from. */
    private transient PCollection<ElemT> pCollection;

    /** A unique tag for the view, typed according to the elements underlying the view. */
    private TupleTag<PrimitiveViewT> tag;

    private WindowMappingFn<W> windowMappingFn;

    /** The windowing strategy for the PCollection underlying the view. */
    private WindowingStrategy<?, W> windowingStrategy;

    /** The coder for the elements underlying the view. */
    private @Nullable Coder<ElemT> coder;

    /** The typed {@link ViewFn} for this view. */
    private ViewFn<PrimitiveViewT, ViewT> viewFn;

    /**
     * Call this constructor to initialize the fields for which this base class provides boilerplate
     * accessors.
     */
    private SimplePCollectionView(
        PCollection<ElemT> pCollection,
        TupleTag<PrimitiveViewT> tag,
        ViewFn<PrimitiveViewT, ViewT> viewFn,
        WindowMappingFn<W> windowMappingFn,
        WindowingStrategy<?, W> windowingStrategy) {
      super(pCollection.getPipeline());
      this.pCollection = pCollection;
      this.windowMappingFn = windowMappingFn;
      this.tag = tag;
      this.windowingStrategy = windowingStrategy;
      this.viewFn = viewFn;
      this.coder = pCollection.getCoder();
    }

    /**
     * Call this constructor to initialize the fields for which this base class provides boilerplate
     * accessors, with an auto-generated tag.
     */
    private SimplePCollectionView(
        PCollection<ElemT> pCollection,
        ViewFn<PrimitiveViewT, ViewT> viewFn,
        WindowMappingFn<W> windowMappingFn,
        WindowingStrategy<?, W> windowingStrategy) {
      this(pCollection, new TupleTag<>(), viewFn, windowMappingFn, windowingStrategy);
    }

    @Override
    public ViewFn<PrimitiveViewT, ViewT> getViewFn() {
      return viewFn;
    }

    @Override
    public WindowMappingFn<?> getWindowMappingFn() {
      return windowMappingFn;
    }

    @Override
    public PCollection<?> getPCollection() {
      return pCollection;
    }

    /**
     * Returns a unique {@link TupleTag} identifying this {@link PCollectionView}.
     *
     * <p>For internal use only by runner implementors.
     */
    @Override
    public TupleTag<?> getTagInternal() {
      return tag;
    }

    /**
     * Returns the {@link WindowingStrategy} of this {@link PCollectionView}, which should be that
     * of the underlying {@link PCollection}.
     *
     * <p>For internal use only by runner implementors.
     */
    @Override
    public WindowingStrategy<?, ?> getWindowingStrategyInternal() {
      return windowingStrategy;
    }

    @Override
    public Coder<?> getCoderInternal() {
      return coder;
    }

    @Override
    public int hashCode() {
      return Objects.hash(tag);
    }

    @Override
    public boolean equals(@Nullable Object other) {
      if (!(other instanceof PCollectionView)) {
        return false;
      }
      @SuppressWarnings("unchecked")
      PCollectionView<?> otherView = (PCollectionView<?>) other;
      return tag.equals(otherView.getTagInternal());
    }

    @Override
    public String toString() {
      return MoreObjects.toStringHelper(this).add("tag", tag).toString();
    }

    @Override
    public Map<TupleTag<?>, PValue> expand() {
      return Collections.singletonMap(tag, pCollection);
    }
  }

  /** A {@link MultimapView} to {@link Map Map<K, V>} adapter. */
  private static class MultimapViewToMapAdapter<K, V> extends AbstractMap<K, V> {
    private final MultimapView<K, V> primitiveViewT;
    private final Supplier<Integer> size;

    private MultimapViewToMapAdapter(MultimapView<K, V> primitiveViewT) {
      this.primitiveViewT = primitiveViewT;
      this.size = Suppliers.memoize(() -> Iterables.size(primitiveViewT.get()));
    }

    @Override
    public boolean containsKey(Object key) {
      return primitiveViewT.get((K) key).iterator().hasNext();
    }

    @Override
    public V get(Object key) {
      Iterator<V> iterator = primitiveViewT.get((K) key).iterator();
      if (!iterator.hasNext()) {
        return null;
      }
      V value = iterator.next();
      if (iterator.hasNext()) {
        throw new IllegalArgumentException("Duplicate values for " + key);
      }
      return value;
    }

    @Override
    public int size() {
      return size.get();
    }

    @Override
    public Set<Entry<K, V>> entrySet() {
      return new AbstractSet<Entry<K, V>>() {
        @Override
        public Iterator<Entry<K, V>> iterator() {
          return FluentIterable.from(primitiveViewT.get())
              .<Entry<K, V>>transform((K key) -> new SimpleEntry<>(key, get(key)))
              .iterator();
        }

        @Override
        public boolean contains(Object o) {
          if (!(o instanceof Entry)) {
            return false;
          }
          Entry<?, ?> entry = (Entry<?, ?>) o;
          // We treat the absence of the key in the map as a difference in these abstract sets. The
          // underlying primitive view represents missing keys as empty iterables so we use this
          // to check if the map contains the key first before comparing values.
          Iterable<V> value = primitiveViewT.get((K) entry.getKey());
          if (value.iterator().hasNext()) {
            return false;
          }
          return Objects.equals(entry.getValue(), value);
        }

        @Override
        public int size() {
          return size.get();
        }
      };
    }
  }

  /** A {@link MultimapView} to {@link Map Map<K, Iterable<V>>} adapter. */
  private static class MultimapViewToMultimapAdapter<K, V> extends AbstractMap<K, Iterable<V>> {
    private final MultimapView<K, V> primitiveViewT;
    private final Supplier<Integer> size;

    private MultimapViewToMultimapAdapter(MultimapView<K, V> primitiveViewT) {
      this.primitiveViewT = primitiveViewT;
      this.size = Suppliers.memoize(() -> Iterables.size(primitiveViewT.get()));
    }

    @Override
    public boolean containsKey(Object key) {
      return primitiveViewT.get((K) key).iterator().hasNext();
    }

    @Override
    public Iterable<V> get(Object key) {
      Iterable<V> values = primitiveViewT.get((K) key);
      // The only way for the values iterable to be empty is for us to have never seen such a key
      // during materialization and hence we return 'null' to satisfy Java's Map contract.
      if (values.iterator().hasNext()) {
        return values;
      } else {
        return null;
      }
    }

    @Override
    public int size() {
      return size.get();
    }

    @Override
    public Set<Entry<K, Iterable<V>>> entrySet() {
      return new AbstractSet<Entry<K, Iterable<V>>>() {
        @Override
        public Iterator<Entry<K, Iterable<V>>> iterator() {
          return FluentIterable.from(primitiveViewT.get())
              .<Entry<K, Iterable<V>>>transform(
                  (K key) -> new SimpleEntry<>(key, primitiveViewT.get(key)))
              .iterator();
        }

        @Override
        public boolean contains(Object o) {
          if (!(o instanceof Entry)) {
            return false;
          }
          Entry<?, ?> entry = (Entry<?, ?>) o;
          if (!(entry.getValue() instanceof Iterable)) {
            return false;
          }
          // We treat the absence of the key in the map as a difference in these abstract sets. The
          // underlying primitive view represents missing keys as empty iterables so we use this
          // to check if the map contains the key first before comparing values.
          Iterable<V> value = primitiveViewT.get((K) entry.getKey());
          if (value.iterator().hasNext()) {
            return false;
          }
          return Iterables.elementsEqual((Iterable<?>) entry.getValue(), value);
        }

        @Override
        public int size() {
          return size.get();
        }
      };
    }
  }
}