PCollectionViews { @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); PriorityQueue<OffsetRange> rangesWithSameFrom = new PriorityQueue<>(OffsetRangeComparator.INSTANCE); Iterator<OffsetRange> iterator = sortedRanges.iterator(); List<OffsetRange> rangesToProcess = new ArrayList<>(); while (iterator.hasNext()) { OffsetRange current = iterator.next(); if (current.getFrom() == current.getTo()) { continue; } 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(); for (; i < rangesToProcess.size(); ++i) { if (rangesToProcess.get(i).getTo() > current.getFrom()) { break; } 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(); } } if (lastTo < current.getFrom() && i != rangesToProcess.size()) { rval.put(new OffsetRange(lastTo, current.getFrom()), rangesToProcess.size() - i); } for (; i < rangesToProcess.size(); ++i) { rangesWithSameFrom.add( new OffsetRange(current.getFrom(), rangesToProcess.get(i).getTo())); } rangesToProcess.clear(); } rangesWithSameFrom.add(current); } while (!rangesWithSameFrom.isEmpty()) { OffsetRange current = rangesWithSameFrom.remove(); rangesToProcess.addAll(rangesWithSameFrom); Collections.sort(rangesToProcess, OffsetRangeComparator.INSTANCE); rangesWithSameFrom.clear(); rval.put(current, rangesToProcess.size() + 1 ); for (OffsetRange rangeWithDifferentFrom : rangesToProcess) { if (rangeWithDifferentFrom.getTo() > current.getTo()) { rangesWithSameFrom.add(new OffsetRange(current.getTo(), rangeWithDifferentFrom.getTo())); } } rangesToProcess.clear(); } return rval.build(); }  static PCollectionView<T> singletonView(
      PCollection<T> pCollection,
      TypeDescriptorSupplier<T> typeDescriptorSupplier,
      WindowingStrategy<?, W> windowingStrategy,
      boolean hasDefault,
      @Nullable T defaultValue,
      Coder<T> defaultValueCoder); @Deprecated static PCollectionView<T> singletonViewUsingVoidKey(
      PCollection<KV<Void, T>> pCollection,
      TypeDescriptorSupplier<T> typeDescriptorSupplier,
      WindowingStrategy<?, W> windowingStrategy,
      boolean hasDefault,
      @Nullable T defaultValue,
      Coder<T> defaultValueCoder); static PCollectionView<Iterable<T>> iterableView(
      PCollection<T> pCollection,
      TypeDescriptorSupplier<T> typeDescriptorSupplier,
      WindowingStrategy<?, W> windowingStrategy); @Deprecated static PCollectionView<Iterable<T>> iterableViewUsingVoidKey(
      PCollection<KV<Void, T>> pCollection,
      TypeDescriptorSupplier<T> typeDescriptorSupplier,
      WindowingStrategy<?, W> windowingStrategy); static PCollectionView<List<T>> listView(
      PCollection<KV<Long, ValueOrMetadata<T, OffsetRange>>> pCollection,
      TypeDescriptorSupplier<T> typeDescriptorSupplier,
      WindowingStrategy<?, W> windowingStrategy); @Deprecated static PCollectionView<List<T>> listViewUsingVoidKey(
      PCollection<KV<Void, T>> pCollection,
      TypeDescriptorSupplier<T> typeDescriptorSupplier,
      WindowingStrategy<?, W> windowingStrategy); static PCollectionView<Map<K, V>> mapView(
      PCollection<KV<K, V>> pCollection,
      TypeDescriptorSupplier<K> keyTypeDescriptorSupplier,
      TypeDescriptorSupplier<V> valueTypeDescriptorSupplier,
      WindowingStrategy<?, W> windowingStrategy); @Deprecated static PCollectionView<Map<K, V>> mapViewUsingVoidKey(
      PCollection<KV<Void, KV<K, V>>> pCollection,
      TypeDescriptorSupplier<K> keyTypeDescriptorSupplier,
      TypeDescriptorSupplier<V> valueTypeDescriptorSupplier,
      WindowingStrategy<?, W> windowingStrategy); static PCollectionView<Map<K, Iterable<V>>> multimapView(
      PCollection<KV<K, V>> pCollection,
      TypeDescriptorSupplier<K> keyTypeDescriptorSupplier,
      TypeDescriptorSupplier<V> valueTypeDescriptorSupplier,
      WindowingStrategy<?, W> windowingStrategy); @Deprecated static PCollectionView<Map<K, Iterable<V>>> multimapViewUsingVoidKey(
          PCollection<KV<Void, KV<K, V>>> pCollection,
          TypeDescriptorSupplier<K> keyTypeDescriptorSupplier,
          TypeDescriptorSupplier<V> valueTypeDescriptorSupplier,
          WindowingStrategy<?, W> windowingStrategy); static Map<TupleTag<?>, PValue> toAdditionalInputs(Iterable<PCollectionView<?>> views);  }
@Test public void testNoOverlapping() { Iterable<OffsetRange> ranges = Arrays.asList(range(0, 2), range(4, 6)); Map<OffsetRange, Integer> nonOverlappingRangesToNumElementsPerPosition = computeOverlappingRanges(ranges); assertEquals( ImmutableMap.of(range(0, 2), 1, range(4, 6), 1), nonOverlappingRangesToNumElementsPerPosition); assertNonEmptyRangesAndPositions(ranges, nonOverlappingRangesToNumElementsPerPosition); }
@Test public void testOnlyTouchingRanges() { Iterable<OffsetRange> ranges = Arrays.asList(range(0, 4), range(4, 8), range(8, 12)); Map<OffsetRange, Integer> nonOverlappingRangesToNumElementsPerPosition = computeOverlappingRanges(ranges); assertEquals( ImmutableMap.of(range(0, 4), 1, range(4, 8), 1, range(8, 12), 1), nonOverlappingRangesToNumElementsPerPosition); assertNonEmptyRangesAndPositions(ranges, nonOverlappingRangesToNumElementsPerPosition); }
@Test public void testRangesWithAtMostOneOverlap() { Iterable<OffsetRange> ranges = Arrays.asList(range(0, 6), range(4, 10), range(8, 12)); Map<OffsetRange, Integer> nonOverlappingRangesToNumElementsPerPosition = computeOverlappingRanges(ranges); assertEquals( ImmutableMap.builder() .put(range(0, 4), 1) .put(range(4, 6), 2) .put(range(6, 8), 1) .put(range(8, 10), 2) .put(range(10, 12), 1) .build(), nonOverlappingRangesToNumElementsPerPosition); assertNonEmptyRangesAndPositions(ranges, nonOverlappingRangesToNumElementsPerPosition); }
@Test public void testOverlappingFroms() { Iterable<OffsetRange> ranges = Arrays.asList(range(0, 4), range(0, 8), range(0, 12)); Map<OffsetRange, Integer> nonOverlappingRangesToNumElementsPerPosition = computeOverlappingRanges(ranges); assertEquals( ImmutableMap.builder().put(range(0, 4), 3).put(range(4, 8), 2).put(range(8, 12), 1).build(), nonOverlappingRangesToNumElementsPerPosition); assertNonEmptyRangesAndPositions(ranges, nonOverlappingRangesToNumElementsPerPosition); }
@Test public void testOverlappingTos() { Iterable<OffsetRange> ranges = Arrays.asList(range(0, 12), range(4, 12), range(8, 12)); Map<OffsetRange, Integer> nonOverlappingRangesToNumElementsPerPosition = computeOverlappingRanges(ranges); assertEquals( ImmutableMap.builder().put(range(0, 4), 1).put(range(4, 8), 2).put(range(8, 12), 3).build(), nonOverlappingRangesToNumElementsPerPosition); assertNonEmptyRangesAndPositions(ranges, nonOverlappingRangesToNumElementsPerPosition); }
@Test public void testOverlappingFromsAndTos() { Iterable<OffsetRange> ranges = Arrays.asList(range(0, 4), range(0, 4), range(0, 4)); Map<OffsetRange, Integer> nonOverlappingRangesToNumElementsPerPosition = computeOverlappingRanges(ranges); assertEquals( ImmutableMap.builder().put(range(0, 4), 3).build(), nonOverlappingRangesToNumElementsPerPosition); assertNonEmptyRangesAndPositions(ranges, nonOverlappingRangesToNumElementsPerPosition); }
@Test public void testMultipleOverlapsForTheSameRange() { Iterable<OffsetRange> ranges = Arrays.asList( range(0, 4), range(0, 8), range(0, 12), range(0, 12), range(4, 12), range(8, 12), range(0, 4), range(0, 8), range(0, 12), range(0, 12), range(4, 12), range(8, 12)); Map<OffsetRange, Integer> nonOverlappingRangesToNumElementsPerPosition = computeOverlappingRanges(ranges); assertEquals( ImmutableMap.builder().put(range(0, 4), 8).put(range(4, 8), 8).put(range(8, 12), 8).build(), nonOverlappingRangesToNumElementsPerPosition); assertNonEmptyRangesAndPositions(ranges, nonOverlappingRangesToNumElementsPerPosition); }
@Test public void testIncreasingOverlaps() { Iterable<OffsetRange> ranges = Arrays.asList(range(0, 4), range(1, 5), range(2, 6), range(3, 7), range(4, 8), range(5, 9)); Map<OffsetRange, Integer> nonOverlappingRangesToNumElementsPerPosition = computeOverlappingRanges(ranges); assertEquals( ImmutableMap.builder() .put(range(0, 1), 1) .put(range(1, 2), 2) .put(range(2, 3), 3) .put(range(3, 4), 4) .put(range(4, 5), 4) .put(range(5, 6), 4) .put(range(6, 7), 3) .put(range(7, 8), 2) .put(range(8, 9), 1) .build(), nonOverlappingRangesToNumElementsPerPosition); assertNonEmptyRangesAndPositions(ranges, nonOverlappingRangesToNumElementsPerPosition); }
@Test public void testNestedOverlaps() { Iterable<OffsetRange> ranges = Arrays.asList(range(0, 8), range(1, 7), range(2, 6), range(3, 5)); Map<OffsetRange, Integer> nonOverlappingRangesToNumElementsPerPosition = computeOverlappingRanges(ranges); assertEquals( ImmutableMap.builder() .put(range(0, 1), 1) .put(range(1, 2), 2) .put(range(2, 3), 3) .put(range(3, 5), 4) .put(range(5, 6), 3) .put(range(6, 7), 2) .put(range(7, 8), 1) .build(), nonOverlappingRangesToNumElementsPerPosition); assertNonEmptyRangesAndPositions(ranges, nonOverlappingRangesToNumElementsPerPosition); }
@Test public void testRandomRanges() { Random random = new Random(123892154890L); for (int i = 0; i < 1000; ++i) { List<OffsetRange> ranges = new ArrayList<>(); for (int j = 0; j < 20; ++j) { long start = random.nextInt(10); ranges.add(range(start, start + random.nextInt(10) + 1)); Map<OffsetRange, Integer> nonOverlappingRangesToNumElementsPerPosition = computeOverlappingRanges(ranges); assertNonEmptyRangesAndPositions(ranges, nonOverlappingRangesToNumElementsPerPosition); } } }