@Override
	public void beforeStep(StepExecution stepExecution) {
		if(StringUtils.hasText(checkColumn)) {

			if(overrideValue != null && overrideValue >= 0) {
				this.incrementalMin = overrideValue;
			}
			else {
				String jobName = stepExecution.getJobExecution().getJobInstance().getJobName();

				// Get the last jobInstance...not the current one
				List<JobInstance> jobInstances = jobExplorer.getJobInstances(jobName, 1, 1);

				if(jobInstances.size() > 0) {
					JobInstance lastInstance = jobInstances.get(jobInstances.size() - 1);

					List<JobExecution> executions = jobExplorer.getJobExecutions(lastInstance);

					JobExecution lastExecution = executions.get(0);

					for (JobExecution execution : executions) {
						if(lastExecution.getEndTime().getTime() < execution.getEndTime().getTime()) {
							lastExecution = execution;
						}
					}

					if(lastExecution.getExecutionContext().containsKey(BATCH_INCREMENTAL_MAX_ID)) {
						this.incrementalMin = lastExecution.getExecutionContext().getLong(BATCH_INCREMENTAL_MAX_ID);
					}
					else {
						this.incrementalMin = Long.MIN_VALUE;
					}
				}
				else {
					this.incrementalMin = Long.MIN_VALUE;
				}
			}

			long newMin = jdbcTemplate.queryForObject(String.format("select max(%s) from %s", checkColumn, table), Integer.class);

			stepExecution.getExecutionContext().put(BATCH_INCREMENTAL_MAX_ID, newMin);
		}

		if(StringUtils.hasText(column) && StringUtils.hasText(table)) {
			if(StringUtils.hasText(checkColumn)) {
				Long minResult = jdbcTemplate.queryForObject("SELECT MIN(" + column + ") from " + table + " where " + checkColumn + " > " + this.incrementalMin, Long.class);
				Long maxResult = jdbcTemplate.queryForObject("SELECT MAX(" + column + ") from " + table + " where " + checkColumn + " > " + this.incrementalMin, Long.class);
				this.partitionMin = minResult != null ? minResult : Long.MIN_VALUE;
				this.partitionMax = maxResult != null ? maxResult : Long.MAX_VALUE;
			}
			else {
				Long minResult = jdbcTemplate.queryForObject("SELECT MIN(" + column + ") from " + table, Long.class);
				Long maxResult = jdbcTemplate.queryForObject("SELECT MAX(" + column + ") from " + table, Long.class);
				this.partitionMin = minResult != null ? minResult : Long.MIN_VALUE;
				this.partitionMax = maxResult != null ? maxResult : Long.MAX_VALUE;
			}
		}
	}
Partition a database table assuming that the data in the column specified
are uniformly distributed. The execution context values will have keys
<code>minValue</code> and <code>maxValue</code> specifying the range of
values to consider in each partition.

@see Partitioner#partition(int)
extensions/spring-xd-extension-batch/src/main/java/org/springframework/batch/integration/x/IncrementalColumnRangePartitioner.java