class EarliestByOffset {

  @VisibleForTesting
  static <T> Udaf<T, Struct, T> earliest(
      final Schema structSchema,
      final boolean ignoreNulls
  ) {
    return new Udaf<T, Struct, T>() {

      @Override
      public Struct initialize() {
        return null;
      }

      @Override
      public Struct aggregate(final T current, final Struct aggregate) {
        if (aggregate != null) {
          return aggregate;
        }

        if (current == null && ignoreNulls) {
          return null;
        }

        return createStruct(structSchema, current);
      }

      @Override
      public Struct merge(final Struct aggOne, final Struct aggTwo) {
        if (aggOne == null) {
          return aggTwo;
        }

        if (aggTwo == null) {
          return aggOne;
        }

        // When merging we need some way of evaluating the "earliest' one.
        // We do this by keeping track of the sequence of when it was originally processed
        if (INTERMEDIATE_STRUCT_COMPARATOR.compare(aggOne, aggTwo) < 0) {
          return aggOne;
        } else {
          return aggTwo;
        }
      }

      @Override
      @SuppressWarnings("unchecked")
      public T map(final Struct agg) {
        if (agg == null) {
          return null;
        }

        return (T) agg.get(VAL_FIELD);
      }
    };
  }

  private  EarliestByOffset();

}