package gw.rating.rtm.matchop

uses gw.api.database.Query
uses gw.rating.rtm.domain.OrderedPersistenceAdapter

uses java.lang.Comparable
uses java.lang.IllegalStateException
uses java.lang.Integer
uses java.math.RoundingMode
uses java.util.RandomAccess
uses java.util.List

@ReadOnly
abstract class StatelessMatchOperator implements ParamConversion {
  protected delegate converter : ParamConversion represents ParamConversion
  var op : RateTableMatchOp as readonly RateTableMatchOp
  var params : List<entity.RateTableColumn> as readonly SortedParams
  var column : String as readonly ColumnName
  var colType : RateTableDataType as readonly ColumnType
  var argIdx : int as readonly ArgIndex
  var merger : SubrangeMerger

  // values for getScore().  getScore() in the default implementation must return Comparable<Object>,
  // but subclasses are free to return compatible (covariant) values.
  protected final var MATCH_SCORE    : Comparable = 1 as Integer
  protected final var NONMATCH_SCORE : Comparable = 0 as Integer

  protected final var ACCEPT_ALL_MIN : int = Integer.MAX_VALUE
  protected final var ACCEPT_ALL_MAX : int = Integer.MIN_VALUE

  construct(matchOp : RateTableMatchOp) {
    this.op = matchOp
    this.params = RateTableMatchOp.sortedParams()
    this.argIdx = matchOp.Definition.SortedMatchOps.indexOf(matchOp)

    var firstParam = SortedParams.first()
    this.column = firstParam.PhysicalColumnName
    this.colType = firstParam.ColumnType
  }

  public property get OrderDependent() : boolean {
    return false
  }

  /**
  * Certain columns will no longer be sortable, specifically those containing multiple
  * values.  Before running comparisons on values that aren't sortable, one should check
  * whether the value is sortable and decide how to deal with the column if not.
   */
  public property get Sortable() : boolean {
    return true
  }
  /**
   * Certain match operations will accept multiple ranges, and need to
   * have the resulting rows re-ordered so that subsequent matches will
   * work on them.
   */
  function needsSubrangeMerge(isNormalized : boolean) : boolean {
    return false
  }

  function getSubrangeMerger(matchOps : List<StatelessMatchOperator>, sortingOrder: int[], isNormalized : boolean) : SubrangeMerger {
    if (needsSubrangeMerge(isNormalized)) {
      if (merger == null) {
        // No lock.  Worst case here is that some garbage will be generated.
        merger = new SubrangeMerger(matchOps, argIdx, sortingOrder)
      }
      return merger
    } else {
      return null
    }
  }

  final function makeConvertedParam(v : Comparable) : Comparable {
    try {
      var n = params.size()
      if (n == 1) {
        return convertParam(v, 0)
      } else if (n == 2) {
        return ConvertedParam.makeConvertedParam(convertParam(v, 0), convertParam(v, 1), v.toString())
      }

      // General case.  Originally this was coded in a functional manner, as
      //     ConvertedParam.makeConvetedParam((0..|n).map(\ i -> convertParam(v, i)).toTypedArray())
      // but the profiler showed both the map and the toTypedArray() calls to be extremely slow.
      var tmp = new Comparable[n]
      for (i in 0..|n) tmp[i] = convertParam(v, i)
      return ConvertedParam.makeConvertedParam(tmp, v.toString())
    } catch (e : java.lang.NullPointerException) {
      if (converter == null) {
        throw new IllegalStateException("MatchOperator " +this+ " was constructed without converter delegate", e)
      }
    }

    return null
  }

  final protected function initConverter() {
    // do this at the end of the subclass constructor, so it has access to a fully-realized MatchOperator
    converter = ParamConverter.makeConverter(this)
  }

  property get MatchOpName() : String {
    return op.Name
  }

  /**
   * get the rounding mode that should be used to convert param #paramIndex
   * from decimal to integer.  paramIndex is zero-based.
   */
  protected function getRounding(paramIndex : int) : RoundingMode {
    return UNNECESSARY
  }

  /**
   * Allow matchop to control the initial sorting of the table rows.
   * @param a The field on one of the rows being sorted
   * @param b The field on the other row being sorted
   * @return a negative integer, zero, or a positive integer as a
   *          is less than, equal to, or greater than b.
   */
  function compareRowValues(a : Comparable, b : Comparable) : int {
    return a.compareTo(b)
  }

  /**
   * Compare the appropriate field in the given bean with the given arg, returning an integer
   * with the semantics of Comparable.compareTo
   *
   * @param bean A rate table row which is being evaluated for match
   * @param arg The match value for this column of the table
   * @return a negative integer, zero, or a positive integer as the bean field
   *          is less than, equal to, or greater than the arg.
   */
   abstract function compare(bean : OrderedPersistenceAdapter, arg : Comparable) : int

  /**
   * Compare the appropriate fields between the two beans and indicate if they overlap.  Most of the
   * time, the values will be directly compared, but will be trickier for range values
   * @param beanA A row to be compared to another
   * @param beanB The row to be compared with
   * @return true if the two rows are representing overlapping or equivalent values
  */
   function overlaps(beanA : OrderedPersistenceAdapter, beanB : OrderedPersistenceAdapter) : boolean {
     final var valA = beanA.get(ArgIndex)
     final var valB = beanB.get(ArgIndex)

     return valA == null ? valB == null : valA.equals(valB)
   }

  /**
   * getScore(bean) is used to break ties when there are multiple matches.   Highest score wins.   This is
   * implemented using list enhancements with orderByDescending().thenByDescending() etc.   Scores
   * are only considered for MatchOperations that have not been relaxed.  see AbstractFactorQuery for how
   * getScore() is used.
   *
   * This implementation of getScore() assumes a single match, and leverages the in-memory filter() method
   * to return a higher score when the filter is satisfied and a lower score when it is not.
   *
   * @param bean A row which satisfies the filter for this MatchOperation
   * @return A Comparable value, where best match has highest value
   */
  abstract function getScore(bean : OrderedPersistenceAdapter, arg : Comparable) : Comparable

  /**
   * The database filter function adds restrictions to the base query so that
   * only rows that satisfy the match op are retained in the result set.
   */
  abstract function filter(query : Query<KeyableBean>, arg : Comparable)

  /**
   * The database relax function adds restrictions to the base query so that
   * only rows that have null in the match op's columns are retained in the result set.
   */
  function relax(query : Query<KeyableBean>) : Query<KeyableBean> {
    this.RateTableMatchOp.sortedParams().each(\ p -> {
      query.compare(p.PhysicalColumnName, Equals, null)
    })
    query.withLogSQL(true)
    return query
  }

  /**
   * The in-memory filter function evaluates all rows still under consideration, and
   * returns only those rows which satisfy the conditions for the matchop.   Filter will
   * only be called when the matchop is active.  Corresponding "relax" is not needed, because
   * the list of results to be filtered will be pre-partitioned according to whether the row
   * would match a filter or a relax.
   *
   * IMPORTANT: the input results are required to be fully ordered...and should stay that way upon output.
   */
  function filter(results : List<OrderedPersistenceAdapter>, arg : Comparable, normalized : boolean) : List<OrderedPersistenceAdapter> {
    // Preliminary tests suggest that the binary search fastFilter is faster even for small N,
    // possibly because of the overhead involved in Gosu list iteration.  But if the list does
    // not support fast random access all bets are off.  In any case, we are required to protect
    // fastFilter from lists with < 2 elements
    var n = results.size()
    if (n == 0) {
      return results
    } else if (n == 1) {
      var r = results.get(0)
      return compare(r, arg) == 0 ? results : {}
    // non-normalized multi select rows are not in a proper sort order, making them ineligible for fast filtering
    } else if (not Sortable and not normalized) {
      return results.where( \ elt -> compare(elt, arg) == 0)
    } else if (results.size() < 3 or not (results typeis RandomAccess)) {
      var s = null as Integer
      var e = 0

      for (r in results index i) {
        if (compare(r, arg) == 0) {
          if (s == null) s = i
          e = i
        }
      }
      return s == null ? results.subList(0, 0) : results.subList(s, e + 1)
    }

    // binary search to find both ends of the range.
    return fastFilter(results, arg)
  }

  // wrapper used to reverse the sense of comparison.
  static class ReverseComparable<C extends Comparable> implements Comparable {
    var _c : C as Value

    construct(c: C) {
      this._c = c
    }

    override function compareTo(p0 : Object) : int {
      return - Value.compareTo((p0 as ReverseComparable<C>).Value)
    }

    override function toString() : String {
      return "ReverseComparable[" + Value + " : " + C.Type.RelativeName + "]"
    }
  }

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

  // Memory filtering via binary search

  // INPUT LIST (results) MUST CONTAIN AT LEAST 2 ELEMENTS.
  // For Greater/Less matchops, override compareMin or compareMax to trigger the "ignore xxx bound" conditions.
  // For interpolated matchop, probably don't even call this...just accept all rows.

  // Note that if you're accepting something more than a narrow range, you probably want to re-order the result based
  // on the downstream match fields only.   See if you can find an O(n) merge that you can do on the subranges...?
  protected function fastFilter(results : List<OrderedPersistenceAdapter>, arg : Comparable) : List<OrderedPersistenceAdapter> {
    var imin = 0
    var imax = results.size() - 1

    do {
      var imid = imin + (imax - imin)/2

      var cmp = compareMid(results.get(imid), arg)
      if (cmp == 0) {
        break
      } else if (cmp < 0) {
        imin = imid + 1
      } else {
        imax = imid
      }
    } while (imin < imax)

    // at this point, results[imin] MIGHT match; results[imax] should not unless imin == imax
    if (imin == imax and compare(results.get(imin), arg) == 0) {
      return results.subList(imin, imax + 1)
    }

    // this loop should give imin as the first element
    var keepMax = imax
    do {
      var imid = imin + (imax - imin)/2

      var cmp = compareMin(results.get(imid), arg)
      if (cmp < 0) {
        imin = imid + 1
      } else if (cmp == ACCEPT_ALL_MIN) {
        // ignore lower bound
        imin = 0
        break
      } else {
        imax = imid
      }
    } while (imin < imax)

    var keepMin = imin

    imax = keepMax

    // this loop should give imax as the last element
    do {
      var imid = imin + (imax - imin + 1)/2
      if (imid > imax) break

      var cmp = compareMax(results.get(imid), arg)
      if (cmp > 0) {
        imax = imid - 1
      } else if (cmp == ACCEPT_ALL_MAX) {
        // ignore upper bound
        imax = results.size() - 1
        break
      } else {
        imin = imid
      }
    } while (imin < imax)

    return results.subList(keepMin, imax + 1)
  }

  // These can be overridden by subclasses that need to control particular behavior
  // They probably MUST be overridden in cases where arg is Compound
  protected function compareMin(bean : OrderedPersistenceAdapter, arg : Comparable) : int {
    return compare(bean, arg)
  }

  protected function compareMid(bean : OrderedPersistenceAdapter, arg : Comparable) : int {
    return compare(bean, arg)
  }

  protected function compareMax(bean : OrderedPersistenceAdapter, arg : Comparable) : int {
    return compare(bean, arg)
  }
}