code: protected long initRankEnd(BaseRow row) throws Exception {
  		if (isConstantRankEnd) {
  			return rankEnd;
  		} else {
  			Long rankEndValue = rankEndState.value();
  			long curRankEnd = rankEndFetcher.apply(row);
  			if (rankEndValue == null) {
  				rankEnd = curRankEnd;
  				rankEndState.update(rankEnd);
  				return rankEnd;
  			} else {
  				rankEnd = rankEndValue;
  				if (rankEnd != curRankEnd) {
  					// increment the invalid counter when the current rank end not equal to previous rank end
  					invalidCounter.inc();
  				}
  				return rankEnd;
  			}
  		}
  	}
docs: Initialize rank end.
  
  @param row input record
  @return rank end
  @throws Exception