/*
 * Copyright (C) 2015 Google Inc.
 *
 * Licensed 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 com.google.cloud.dataflow.sdk.transforms;

import com.google.cloud.dataflow.sdk.values.KV;
import com.google.cloud.dataflow.sdk.values.PCollection;

/**
 * {@code Count<T>} takes a {@code PCollection<T>} and returns a
 * {@code PCollection<KV<T, Long>>} representing a map from each
 * distinct element of the input {@code PCollection} to the number of times
 * that element occurs in the input.  Each of the keys in the output
 * {@code PCollection} is unique.
 *
 * <p> Two values of type {@code T} are compared for equality <b>not</b> by
 * regular Java {@link Object#equals}, but instead by first encoding
 * each of the elements using the {@code PCollection}'s {@code Coder}, and then
 * comparing the encoded bytes.  This admits efficient parallel
 * evaluation.
 *
 * <p> By default, the {@code Coder} of the keys of the output
 * {@code PCollection} is the same as the {@code Coder} of the
 * elements of the input {@code PCollection}.
 *
 * <p> Each output element is in the window by which its corresponding input
 * was grouped, and has the timestamp of the end of that window.  The output
 * {@code PCollection} has the same
 * {@link com.google.cloud.dataflow.sdk.transforms.windowing.WindowFn}
 * as the input.
 *
 * <p> Example of use:
 * <pre> {@code
 * PCollection<String> words = ...;
 * PCollection<KV<String, Long>> wordCounts =
 *     words.apply(Count.<String>perElement());
 * } </pre>
 */
public class Count {

  /**
   * Returns a {@link Globally Count.Globally} {@link PTransform}
   * that counts the number of elements in its input {@link PCollection}.
   *
   * <p> See {@link Globally Count.Globally} for more details.
   */
  public static <T> Globally<T> globally() {
    return new Globally<>();
  }

  /**
   * Returns a {@link PerElement Count.PerElement} {@link PTransform}
   * that counts the number of occurrences of each element in its
   * input {@link PCollection}.
   *
   * <p> See {@link PerElement Count.PerElement} for more details.
   */
  public static <T> PerElement<T> perElement() {
    return new PerElement<>();
  }

  ///////////////////////////////////////

  /**
   * {@code Count.Globally<T>} takes a {@code PCollection<T>} and returns a
   * {@code PCollection<Long>} containing a single element which is the total
   * number of elements in the {@code PCollection}.
   *
   * <p> Example of use:
   * <pre> {@code
   * PCollection<String> words = ...;
   * PCollection<Long> wordCount =
   *     words.apply(Count.<String>globally());
   * } </pre>
   *
   * @param <T> the type of the elements of the input {@code PCollection}
   */
  @SuppressWarnings("serial")
  public static class Globally<T>
      extends PTransform<PCollection<T>, PCollection<Long>> {

    public Globally() { }

    @Override
    public PCollection<Long> apply(PCollection<T> input) {
      return
          input
          .apply(ParDo.named("Init")
                 .of(new DoFn<T, Long>() {
                     @Override
                     public void processElement(ProcessContext c) {
                       c.output(1L);
                     }
                   }))
          .apply(Sum.longsGlobally());
    }
  }

  /**
   * {@code Count.PerElement<T>} takes a {@code PCollection<T>} and returns a
   * {@code PCollection<KV<T, Long>>} representing a map from each
   * distinct element of the input {@code PCollection} to the number of times
   * that element occurs in the input.  Each of the keys in the output
   * {@code PCollection} is unique.
   *
   * <p> This transform compares two values of type {@code T} by first
   * encoding each element using the input {@code PCollection}'s
   * {@code Coder}, then comparing the encoded bytes. Because of this,
   * the input coder must be deterministic. (See
   * {@link com.google.cloud.dataflow.sdk.coders.Coder#verifyDeterministic()} for more detail).
   * Performing the comparison in this manner admits efficient parallel evaluation.
   *
   * <p> By default, the {@code Coder} of the keys of the output
   * {@code PCollection} is the same as the {@code Coder} of the
   * elements of the input {@code PCollection}.
   *
   * <p> Example of use:
   * <pre> {@code
   * PCollection<String> words = ...;
   * PCollection<KV<String, Long>> wordCounts =
   *     words.apply(Count.<String>perElement());
   * } </pre>
   *
   * @param <T> the type of the elements of the input {@code PCollection}, and
   * the type of the keys of the output {@code PCollection}
   */
  @SuppressWarnings("serial")
  public static class PerElement<T>
      extends PTransform<PCollection<T>, PCollection<KV<T, Long>>> {

    public PerElement() { }

    @Override
    public PCollection<KV<T, Long>> apply(PCollection<T> input) {
      return
          input
          .apply(ParDo.named("Init")
                 .of(new DoFn<T, KV<T, Long>>() {
                     @Override
                     public void processElement(ProcessContext c) {
                       c.output(KV.of(c.element(), 1L));
                     }
                   }))
          .apply(Sum.<T>longsPerKey());
    }

    @Override
    public String getKindString() {
      return "Count.PerElement";
    }
  }
}

@@ -757,1260 +757,322 @@
 %0A * 
-%7B@code Count%3CT%3E%7D takes a %7B@code PCollection%3CT%3E%7D and returns a%0A * %7B@code PCollection%3CKV%3CT, Long%3E%3E%7D representing a map from each%0A * distinct element of the input %7B@code PCollection%7D to the number of times%0A * that element occurs in the input.  Each of the keys in the output%0A * %7B@code PCollection%7D is unique.%0A *%0A * %3Cp%3E Two values of type %7B@code T%7D are compared for equality %3Cb%3Enot%3C/b%3E by%0A * regular Java %7B@link Object#equals%7D, but instead by first encoding%0A * each of the elements using the %7B@code PCollection%7D's %7B@code Coder%7D, and then%0A * comparing the encoded bytes.  This admits efficient parallel%0A * evaluation.%0A *%0A * %3Cp%3E By default, the %7B@code Coder%7D of the keys of the output%0A * %7B@code PCollection%7D is the same as the %7B@code Coder%7D of the%0A * elements of the input %7B@code PCollection%7D.%0A *%0A * %3Cp%3E Each output element is in the window by which its corresponding input%0A * was grouped, and has the timestamp of the end of that window.  The output%0A * %7B@code PCollection%7D has the same%0A * %7B@link com.google.cloud.dataflow.sdk.transforms.windowing.WindowFn%7D%0A * as the input.%0A *%0A * %3Cp%3E Example of use:%0A * %3Cpre%3E %7B@code%0A * PCollection%3CString%3E words = ...;%0A * PCollection%3CKV%3CString, Long%3E%3E wordCounts =%0A *     words.apply(Count.%3CString%3EperElement());%0A * %7D %3C/pre%3E
+Count transforms can be used to count the number of elements in a PCollection.%0A * %7B@link PerElement Count.PerElement can be used to count the number of occurrences of each%0A * distinct element in the PCollection. %7B@link Globally Count.Globally%7D can%0A * be used to count the total number of elements in a PCollection.
 %0A */

Fix the javadoc for Count transforms. ----Release Notes---- [] ------------- Created by MOE: http://code.google.com/p/moe-java MOE_MIGRATED_REVID=88241545