<commit_before>package com.spotify.cloud.dataflow.values

import com.google.cloud.dataflow.sdk.transforms.{DoFn, ParDo}
import com.google.cloud.dataflow.sdk.values.{PCollection, PCollectionView}
import com.spotify.cloud.dataflow.DataflowContext
import com.spotify.cloud.dataflow.util.FunctionsWithSideInput

import scala.reflect.ClassTag

sealed trait SCollectionWithSideInput[T, S] {

  def filter(f: (T, S) => Boolean): SCollectionWithSideInput[T, S]

  def flatMap[U: ClassTag](f: (T, S) => TraversableOnce[U]): SCollectionWithSideInput[U, S]

  def keyBy[K: ClassTag](f: (T, S) => K): SCollectionWithSideInput[(K, T), S]

  def map[U: ClassTag](f: (T, S) => U): SCollectionWithSideInput[U, S]

  def toSCollection: SCollection[T]

}

private class SCollectionWithSideInputImpl[T, S, R](val internal: PCollection[T],
                                                        view: PCollectionView[R],
                                                        sideFn: R => S)
                                                       (implicit val context: DataflowContext, val ct: ClassTag[T])
  extends SCollectionWithSideInput[T, S] with PCollectionWrapper[T] {

  override protected def parDo[U: ClassManifest](fn: DoFn[T, U]): SCollection[U] =
    this.apply(ParDo.withSideInputs(view).of(fn)).setCoder(this.getCoder[U])

  def filter(f: (T, S) => Boolean): SCollectionWithSideInput[T, S] = {
    val o = this.parDo(FunctionsWithSideInput.filterFn(view, sideFn, f)).internal
    new SCollectionWithSideInputImpl[T, S, R](o, view, sideFn)
  }

  def flatMap[U: ClassTag](f: (T, S) => TraversableOnce[U]): SCollectionWithSideInput[U, S] = {
    val o = this.parDo(FunctionsWithSideInput.flatMapFn(view, sideFn, f)).internal
    new SCollectionWithSideInputImpl[U, S, R](o, view, sideFn)
  }

  def keyBy[K: ClassTag](f: (T, S) => K): SCollectionWithSideInput[(K, T), S] = this.map((v, c) => (f(v, c), v))

  def map[U: ClassTag](f: (T, S) => U): SCollectionWithSideInput[U, S] = {
    val o = this.parDo(FunctionsWithSideInput.mapFn(view, sideFn, f)).internal
    new SCollectionWithSideInputImpl[U, S, R](o, view, sideFn)
  }

  def toSCollection: SCollection[T] = SCollection(internal)

}<commit_msg>fix deprecated ClassManifest context bound<commit_after>package com.spotify.cloud.dataflow.values

import com.google.cloud.dataflow.sdk.transforms.{DoFn, ParDo}
import com.google.cloud.dataflow.sdk.values.{PCollection, PCollectionView}
import com.spotify.cloud.dataflow.DataflowContext
import com.spotify.cloud.dataflow.util.FunctionsWithSideInput

import scala.reflect.ClassTag

sealed trait SCollectionWithSideInput[T, S] {

  def filter(f: (T, S) => Boolean): SCollectionWithSideInput[T, S]

  def flatMap[U: ClassTag](f: (T, S) => TraversableOnce[U]): SCollectionWithSideInput[U, S]

  def keyBy[K: ClassTag](f: (T, S) => K): SCollectionWithSideInput[(K, T), S]

  def map[U: ClassTag](f: (T, S) => U): SCollectionWithSideInput[U, S]

  def toSCollection: SCollection[T]

}

private class SCollectionWithSideInputImpl[T, S, R](val internal: PCollection[T],
                                                        view: PCollectionView[R],
                                                        sideFn: R => S)
                                                       (implicit val context: DataflowContext, val ct: ClassTag[T])
  extends SCollectionWithSideInput[T, S] with PCollectionWrapper[T] {

  override protected def parDo[U: ClassTag](fn: DoFn[T, U]): SCollection[U] =
    this.apply(ParDo.withSideInputs(view).of(fn)).setCoder(this.getCoder[U])

  def filter(f: (T, S) => Boolean): SCollectionWithSideInput[T, S] = {
    val o = this.parDo(FunctionsWithSideInput.filterFn(view, sideFn, f)).internal
    new SCollectionWithSideInputImpl[T, S, R](o, view, sideFn)
  }

  def flatMap[U: ClassTag](f: (T, S) => TraversableOnce[U]): SCollectionWithSideInput[U, S] = {
    val o = this.parDo(FunctionsWithSideInput.flatMapFn(view, sideFn, f)).internal
    new SCollectionWithSideInputImpl[U, S, R](o, view, sideFn)
  }

  def keyBy[K: ClassTag](f: (T, S) => K): SCollectionWithSideInput[(K, T), S] = this.map((v, c) => (f(v, c), v))

  def map[U: ClassTag](f: (T, S) => U): SCollectionWithSideInput[U, S] = {
    val o = this.parDo(FunctionsWithSideInput.mapFn(view, sideFn, f)).internal
    new SCollectionWithSideInputImpl[U, S, R](o, view, sideFn)
  }

  def toSCollection: SCollection[T] = SCollection(internal)

}
{'fraction_non_alphanumeric': 0.14059674502712477, 'fraction_numerical': 0.0, 'mean_word_length': 3.6628029504741835, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4631130', 'n_tokens_mistral': 1501, 'n_tokens_neox': 1476, 'n_words': 324}