@Test public void testDataPullUpperBoundsNotRemovedInLastWorkUnit() { int totalCount = 5; ArrayList<DataRecord> records = this.generateRecords(totalCount); WorkUnit workUnit = WorkUnit.createEmpty(); WorkUnitState workUnitState = new WorkUnitState(workUnit, new State()); workUnitState.setId("testDataPullUpperBoundsNotRemovedInLastWorkUnit"); TestQueryBasedExtractor testExtractor = new TestQueryBasedExtractor(workUnitState, records); testExtractor.setRangePredicates(1, 3); this.verify(testExtractor, 3); workUnit.setProp(Partition.IS_LAST_PARTIITON, true); workUnit.setProp(Partition.HAS_USER_SPECIFIED_HIGH_WATERMARK, true); testExtractor.reset(); testExtractor.setRangePredicates(1, 3); this.verify(testExtractor, 3); workUnit.removeProp(Partition.HAS_USER_SPECIFIED_HIGH_WATERMARK); workUnit.setProp(ConfigurationKeys.WORK_UNIT_STATE_ACTUAL_HIGH_WATER_MARK_KEY, "3"); testExtractor.reset(); testExtractor.setRangePredicates(1, 3); this.verify(testExtractor, 3); }
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)); } } }
QueryBasedExtractor implements Extractor<S, D>, ProtocolSpecificLayer<S, D> { 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)); } } } }
QueryBasedExtractor implements Extractor<S, D>, ProtocolSpecificLayer<S, D> { 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)); } } } QueryBasedExtractor(WorkUnitState workUnitState); }
QueryBasedExtractor implements Extractor<S, D>, ProtocolSpecificLayer<S, D> { 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)); } } } QueryBasedExtractor(WorkUnitState workUnitState); boolean getFetchStatus(); void setFetchStatus(boolean fetchStatus); void setHighWatermark(long highWatermark); @Override D readRecord(@Deprecated D reuse); @Override long getExpectedRecordCount(); @Override S getSchema(); @Override long getHighWatermark(); @Override void close(); boolean isFullDump(); Extractor<S, D> build(); }
QueryBasedExtractor implements Extractor<S, D>, ProtocolSpecificLayer<S, D> { 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)); } } } QueryBasedExtractor(WorkUnitState workUnitState); boolean getFetchStatus(); void setFetchStatus(boolean fetchStatus); void setHighWatermark(long highWatermark); @Override D readRecord(@Deprecated D reuse); @Override long getExpectedRecordCount(); @Override S getSchema(); @Override long getHighWatermark(); @Override void close(); boolean isFullDump(); Extractor<S, D> build();  }