code: private void setRangePredicates(String watermarkColumn, WatermarkType watermarkType, long lwmValue, long hwmValue) {
      log.debug("Getting range predicates");
      String lwmOperator = partition.isLowWatermarkInclusive() ? ">=" : ">";
      String hwmOperator = (partition.isLastPartition() || partition.isHighWatermarkInclusive()) ? "<=" : "<";
  
      WatermarkPredicate watermark = new WatermarkPredicate(watermarkColumn, watermarkType);
      this.addPredicates(watermark.getPredicate(this, lwmValue, lwmOperator, Predicate.PredicateType.LWM));
      this.addPredicates(watermark.getPredicate(this, hwmValue, hwmOperator, Predicate.PredicateType.HWM));
  
      if (Boolean.valueOf(this.workUnitState.getProp(ConfigurationKeys.SOURCE_QUERYBASED_IS_HOURLY_EXTRACT))) {
        String hourColumn = this.workUnitState.getProp(ConfigurationKeys.SOURCE_QUERYBASED_HOUR_COLUMN);
        if (StringUtils.isNotBlank(hourColumn)) {
          WatermarkPredicate hourlyWatermark = new WatermarkPredicate(hourColumn, WatermarkType.HOUR);
          this.addPredicates(hourlyWatermark.getPredicate(this, lwmValue, lwmOperator, Predicate.PredicateType.LWM));
          this.addPredicates(hourlyWatermark.getPredicate(this, hwmValue, hwmOperator, Predicate.PredicateType.HWM));
        }
      }
    }
docs: range predicates for watermark column and transaction columns.
  
  @param watermarkColumn name of the column used as watermark
  @param watermarkType watermark type
  @param lwmValue estimated low watermark value
  @param hwmValue estimated high watermark value