package gw.rating.rtm.matchop

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

uses java.lang.Comparable
uses java.lang.IllegalArgumentException
uses java.util.ArrayList
uses java.util.Collections
uses java.util.List

@ReadOnly
abstract class StatelessRangeMatch extends StatelessMatchOperator {
  var minColumn : String
  var maxColumn : String

  construct(matchOp : RateTableMatchOp) {
    super(matchOp)
    minColumn = this.SortedParams[0].PhysicalColumnName
    maxColumn = this.SortedParams[1].PhysicalColumnName
  }


  override function compare(bean: OrderedPersistenceAdapter, arg: Comparable <java.lang.Object>): int {
    var b = getRange(bean)

    var param = arg as CompoundParam
    if (param.First == null or param.Second == null) {
      throw new IllegalArgumentException("Param should not contain null values: [" + param.First + "," + param.Second + "]")
    }

    // bean can contain a null value, but arg cannot.
    if ( b.First != null and atMinCutoff(  b.First.compareTo(param.First)  )) {
      return 1
    } else if ( b.Second != null and atMaxCutoff( b.Second.compareTo(param.Second) )) {
      return -1
    } else {
      return 0
    }
  }

  /**
   * To compare two ranges for overlap, we look to see if the second of each is within the
   * limit of the first of the other.  Likewise, we must deal with the cases that one or
   * both ranges are open ended.  For ease of implementation, we check these first, prevening
   * NPEs
  */
  override function overlaps(beanA : OrderedPersistenceAdapter, beanB : OrderedPersistenceAdapter) : boolean {
    var a = getRange(beanA)
    var b = getRange(beanB)

    if (a.First == null) {
      return a.Second == null
          or b.First == null
          or !atMaxCutoff(a.Second.compareTo(b.First))
    }
    if (a.Second == null) {
      return b.Second == null
          or !atMaxCutoff(b.Second.compareTo(a.First))
    }
    if (b.First == null) {
      return b.Second == null
          or !atMaxCutoff(b.Second.compareTo(a.First))
    }
    if (b.Second == null) {
      return !atMaxCutoff(a.Second.compareTo(b.First))
    }
    return !atMaxCutoff(a.Second.compareTo(b.First))
        and !atMaxCutoff(b.Second.compareTo(a.First))
  }

  // TODO: figure out whether this is necessary.
  override function needsSubrangeMerge(isNormalized : boolean) : boolean {
    return !isNormalized
  }

  override function getScore(bean : OrderedPersistenceAdapter, arg : Comparable) : Comparable {
    return compare(bean, arg) == 0 ? MATCH_SCORE : NONMATCH_SCORE
  }

  override function filter(results : List<OrderedPersistenceAdapter>, arg : Comparable, normalized : boolean) : List<OrderedPersistenceAdapter> {
    if (normalized) {
      return super.filter(results, arg, normalized)
    }

    // If ranges are not normalized, we need to scan every row that could match
    var rows = new ArrayList<OrderedPersistenceAdapter>()
    for (row in results index i) {
      var range = getRange(row)
      if (range.First != null and range.First > (arg as RangeParam ).First) {
        // We've passed all elements that could encompass arg
        break
      } else if (compare(row, arg) == 0) {
        rows.add(row)
      }
    }
    return Collections.unmodifiableList(rows)
  }

  override function filter(query: Query <entity.KeyableBean>, arg: Comparable) {
    if (arg typeis CompoundParam) {
      query.and(\ a -> {
        a.or(\ o -> o.compare(minColumn, Equals,    null).compare(minColumn, MinOp,     arg.First))
        a.or(\ o -> o.compare(maxColumn, Equals,    null).compare(maxColumn, MaxOp,     arg.Second))
        a.or(\ o -> o.compare(minColumn, NotEquals, null).compare(maxColumn, NotEquals, null))
      })
    } else {
      // arg isn't a CompoundParam; probably it hasn't been properly adjusted
      throw new IllegalArgumentException("arg must be a compound param.  Perhaps it wasn't converted properly?")
    }
  }

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

  // used by filter() above
  abstract protected property get MaxOp() : Relop

  protected property get MinOp() : Relop {
    return LessThanOrEquals
  }

  // edge comparisons ued by compare above and compareMax below
  abstract protected function atMaxCutoff(max : int) : boolean

  protected function atMinCutoff(min : int) : boolean {
    return (min > 0)
  }

  // NOTE THAT SUBRANGE FILTERING DOES NOT WORK IF THERE IS A RANGE MATCHOP WITH NONUNIFORM RANGE DATA.
  // For example, if the table contains ranges {1,2},{2,3},{3,5},{5,8},{8,null} things should work OK
  // because there are no ranges that subsume one another.   But {1,2},{2,3},{3,5},{5,8},{8,null}, {1,5},{5,null}
  // will not work correctly because {1,5} overlaps {1,2}{2,3}{3,5} and {5,null} overlaps {5,8}{8,null}
  // There are a number of strategies to consider.
  //    1) If there is a HIGHER-PRIORITY matchop which separates things out (e.g. the granular ranges above
  //       are for Ohio, and the others are for Pennsylvania) things should work OK.
  //    2) Write a partitioner that automatically splits the rows so the granularity is uniform
  //    3) Build a different indexing structure that handles this correctly
  //    4) Run the binary search to get the outside bounds of the range, then run .where(...) on the bounded range

  // TODO: for #4, we could change compareMin to accept everything back to element 0, and use the version of fastFilter below.
  // But this would not perform well, so perhaps we would want to do it only when the granularity is known to cause problems.
  // (i.e. pass an additional parameter that tells whether the rows are uniform)
  // For performance testing, we will instead create data where ranges are uniformly granular.
  //  override protected function fastFilter(results : List<OrderedPersistenceAdapter>, arg : Comparable) : List<OrderedPersistenceAdapter> {
  //    var res = super.fastFilter(results, arg)
  //    return res.where( \ elt -> compare(elt, arg) == 0)
  //  }

  // These are for controlling super.fastFilter, as used above

  override protected function compareMin(bean : OrderedPersistenceAdapter, arg : Comparable) : int {
    // Return "equal" until we encounter a bean where Second is less/lesseq the value of arg.
    var b = getRange(bean)
    return b.Second == null ? 0 : atMaxCutoff( b.Second.compareTo((arg as CompoundParam).Second) ) ? -1 : 0
  }

  override protected function compareMax(bean : OrderedPersistenceAdapter, arg : Comparable) : int {
    // Return "equal" until we encounter a bean where First exceeds the value of arg.
    var b = getRange(bean)
    return b.First == null ? 0 : atMinCutoff( b.First.compareTo((arg as CompoundParam).First) ) ? 1 : 0
  }

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

  // utility for getting the correct param out of a bean

  private function getRange(bean: OrderedPersistenceAdapter) : CompoundParam {
    return bean.get(this.ArgIndex) as CompoundParam
  }
}