Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

IN PROGRESS - BATCH-774: Split item count into read/write/filter

Browse Source 
made read/filter/write counts persistent
This commit is contained in:
robokaso
2008-09-03 12:32:37 +00:00
parent d5f7a1ad48
commit 740d299c80
11 changed files with 72 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

69
spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcStepExecutionDao.java
View File

@@ -41,19 +41,19 @@ public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implement
private static final Log logger = LogFactory.getLog(JdbcStepExecutionDao.class);
private static final String SAVE_STEP_EXECUTION = "INSERT into %PREFIX%STEP_EXECUTION(STEP_EXECUTION_ID, VERSION, STEP_NAME, JOB_EXECUTION_ID, START_TIME, "
+ "END_TIME, STATUS, COMMIT_COUNT, ITEM_COUNT, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE, READ_SKIP_COUNT, WRITE_SKIP_COUNT, ROLLBACK_COUNT, LAST_UPDATED) "
+ "values(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
+ "END_TIME, STATUS, COMMIT_COUNT, READ_COUNT, FILTER_COUNT, WRITE_COUNT, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE, READ_SKIP_COUNT, WRITE_SKIP_COUNT, ROLLBACK_COUNT, LAST_UPDATED) "
+ "values(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
private static final String UPDATE_STEP_EXECUTION = "UPDATE %PREFIX%STEP_EXECUTION set START_TIME = ?, END_TIME = ?, "
+ "STATUS = ?, COMMIT_COUNT = ?, ITEM_COUNT = ?, CONTINUABLE = ? , EXIT_CODE = ?, "
+ "EXIT_MESSAGE = ?, VERSION = ?, READ_SKIP_COUNT = ?, WRITE_SKIP_COUNT = ?, ROLLBACK_COUNT = ?, LAST_UPDATED = ?" +
" where STEP_EXECUTION_ID = ? and VERSION = ?";
+ "STATUS = ?, COMMIT_COUNT = ?, READ_COUNT = ?, FILTER_COUNT = ?, WRITE_COUNT = ?, CONTINUABLE = ? , EXIT_CODE = ?, "
+ "EXIT_MESSAGE = ?, VERSION = ?, READ_SKIP_COUNT = ?, WRITE_SKIP_COUNT = ?, ROLLBACK_COUNT = ?, LAST_UPDATED = ?"
+ " where STEP_EXECUTION_ID = ? and VERSION = ?";
private static final String GET_RAW_STEP_EXECUTIONS = "SELECT STEP_EXECUTION_ID, STEP_NAME, START_TIME, END_TIME, STATUS, COMMIT_COUNT,"
+ " ITEM_COUNT, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE, READ_SKIP_COUNT, WRITE_SKIP_COUNT, ROLLBACK_COUNT, LAST_UPDATED from %PREFIX%STEP_EXECUTION where JOB_EXECUTION_ID = ?";
+ " READ_COUNT, FILTER_COUNT, WRITE_COUNT, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE, READ_SKIP_COUNT, WRITE_SKIP_COUNT, ROLLBACK_COUNT, LAST_UPDATED from %PREFIX%STEP_EXECUTION where JOB_EXECUTION_ID = ?";
private static final String GET_STEP_EXECUTIONS = GET_RAW_STEP_EXECUTIONS + " order by STEP_EXECUTION_ID";
private static final String GET_STEP_EXECUTION = GET_RAW_STEP_EXECUTIONS + " and STEP_NAME = ?";
private static final String CURRENT_VERSION_STEP_EXECUTION = "SELECT VERSION FROM %PREFIX%STEP_EXECUTION WHERE STEP_EXECUTION_ID=?";
@@ -105,15 +105,17 @@ public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implement
Object[] parameters = new Object[] { stepExecution.getId(), stepExecution.getVersion(),
stepExecution.getStepName(), stepExecution.getJobExecutionId(), stepExecution.getStartTime(),
stepExecution.getEndTime(), stepExecution.getStatus().toString(), stepExecution.getCommitCount(),
stepExecution.getItemCount(), stepExecution.getExitStatus().isContinuable() ? "Y" : "N",
stepExecution.getExitStatus().getExitCode(), exitDescription, stepExecution.getReadSkipCount(),
stepExecution.getWriteSkipCount(), stepExecution.getRollbackCount(), stepExecution.getLastUpdated() };
stepExecution.getReadCount(), stepExecution.getFilterCount(), stepExecution.getWriteCount(),
stepExecution.getExitStatus().isContinuable() ? "Y" : "N", stepExecution.getExitStatus().getExitCode(),
exitDescription, stepExecution.getReadSkipCount(), stepExecution.getWriteSkipCount(),
stepExecution.getRollbackCount(), stepExecution.getLastUpdated() };
getJdbcTemplate().getJdbcOperations().update(
getQuery(SAVE_STEP_EXECUTION),
parameters,
new int[] { Types.INTEGER, Types.INTEGER, Types.VARCHAR, Types.INTEGER, Types.TIMESTAMP,
Types.TIMESTAMP, Types.VARCHAR, Types.INTEGER, Types.INTEGER, Types.CHAR, Types.VARCHAR,
Types.VARCHAR, Types.INTEGER, Types.INTEGER, Types.INTEGER, Types.TIMESTAMP });
Types.TIMESTAMP, Types.VARCHAR, Types.INTEGER, Types.INTEGER, Types.INTEGER, Types.INTEGER,
Types.CHAR, Types.VARCHAR, Types.VARCHAR, Types.INTEGER, Types.INTEGER, Types.INTEGER,
Types.TIMESTAMP });
}
/**
@@ -132,7 +134,9 @@ public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implement
/*
* (non-Javadoc)
* @see org.springframework.batch.execution.repository.dao.StepExecutionDao#updateStepExecution(org.springframework.batch.core.domain.StepExecution)
*
* @seeorg.springframework.batch.execution.repository.dao.StepExecutionDao#
* updateStepExecution(org.springframework.batch.core.domain.StepExecution)
*/
public void updateStepExecution(StepExecution stepExecution) {
@@ -151,18 +155,21 @@ public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implement
Integer version = new Integer(stepExecution.getVersion().intValue() + 1);
Object[] parameters = new Object[] { stepExecution.getStartTime(), stepExecution.getEndTime(),
stepExecution.getStatus().toString(), stepExecution.getCommitCount(), stepExecution.getItemCount(),
stepExecution.getStatus().toString(), stepExecution.getCommitCount(), stepExecution.getReadCount(),
stepExecution.getFilterCount(), stepExecution.getWriteCount(),
stepExecution.getExitStatus().isContinuable() ? "Y" : "N",
stepExecution.getExitStatus().getExitCode(), exitDescription, version,
stepExecution.getReadSkipCount(), stepExecution.getWriteSkipCount(),
stepExecution.getRollbackCount(), stepExecution.getLastUpdated(), stepExecution.getId(),
stepExecution.getRollbackCount(), stepExecution.getLastUpdated(), stepExecution.getId(),
stepExecution.getVersion() };
int count = getJdbcTemplate().getJdbcOperations().update(
getQuery(UPDATE_STEP_EXECUTION),
parameters,
new int[] { Types.TIMESTAMP, Types.TIMESTAMP, Types.VARCHAR, Types.INTEGER, Types.INTEGER,
Types.CHAR, Types.VARCHAR, Types.VARCHAR, Types.INTEGER, Types.INTEGER, Types.INTEGER,
Types.INTEGER, Types.TIMESTAMP, Types.INTEGER, Types.INTEGER });
int count = getJdbcTemplate().getJdbcOperations()
.update(
getQuery(UPDATE_STEP_EXECUTION),
parameters,
new int[] { Types.TIMESTAMP, Types.TIMESTAMP, Types.VARCHAR, Types.INTEGER, Types.INTEGER,
Types.INTEGER, Types.INTEGER, Types.CHAR, Types.VARCHAR, Types.VARCHAR,
Types.INTEGER, Types.INTEGER, Types.INTEGER, Types.INTEGER, Types.TIMESTAMP,
Types.INTEGER, Types.INTEGER });
// Avoid concurrent modifications...
if (count == 0) {
@@ -207,13 +214,13 @@ public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implement
return (StepExecution) executions.get(0);
}
}
public List<StepExecution> getStepExecutions(JobExecution jobExecution) {
List<StepExecution> executions = getJdbcTemplate().query(getQuery(GET_STEP_EXECUTIONS),
new StepExecutionRowMapper(jobExecution), jobExecution.getId());
return executions;
}
private class StepExecutionRowMapper implements ParameterizedRowMapper<StepExecution> {
private final JobExecution jobExecution;
@@ -228,15 +235,17 @@ public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implement
stepExecution.setEndTime(rs.getTimestamp(4));
stepExecution.setStatus(BatchStatus.valueOf(rs.getString(5)));
stepExecution.setCommitCount(rs.getInt(6));
stepExecution.setItemCount(rs.getInt(7));
stepExecution.setExitStatus(new ExitStatus("Y".equals(rs.getString(8)), rs.getString(9), rs.getString(10)));
stepExecution.setReadSkipCount(rs.getInt(11));
stepExecution.setWriteSkipCount(rs.getInt(12));
stepExecution.setRollbackCount(rs.getInt(13));
stepExecution.setLastUpdated(rs.getTimestamp(14));
stepExecution.setReadCount(rs.getInt(7));
stepExecution.setFilterCount(rs.getInt(8));
stepExecution.setWriteCount(rs.getInt(9));
stepExecution.setExitStatus(new ExitStatus("Y".equals(rs.getString(10)), rs.getString(11), rs.getString(12)));
stepExecution.setReadSkipCount(rs.getInt(13));
stepExecution.setWriteSkipCount(rs.getInt(14));
stepExecution.setRollbackCount(rs.getInt(15));
stepExecution.setLastUpdated(rs.getTimestamp(16));
return stepExecution;
}
}
}

4
spring-batch-core/src/main/resources/schema-db2.sql
View File

@@ -54,7 +54,9 @@ CREATE TABLE BATCH_STEP_EXECUTION (
END_TIME TIMESTAMP DEFAULT NULL ,
STATUS VARCHAR(10) ,
COMMIT_COUNT BIGINT ,
ITEM_COUNT BIGINT ,
READ_COUNT BIGINT ,
FILTER_COUNT BIGINT ,
WRITE_COUNT BIGINT ,
READ_SKIP_COUNT BIGINT ,
WRITE_SKIP_COUNT BIGINT ,
ROLLBACK_COUNT BIGINT ,

4
spring-batch-core/src/main/resources/schema-derby.sql
View File

@@ -54,7 +54,9 @@ CREATE TABLE BATCH_STEP_EXECUTION (
END_TIME TIMESTAMP DEFAULT NULL ,
STATUS VARCHAR(10) ,
COMMIT_COUNT BIGINT ,
ITEM_COUNT BIGINT ,
READ_COUNT BIGINT ,
FILTER_COUNT BIGINT ,
WRITE_COUNT BIGINT ,
READ_SKIP_COUNT BIGINT ,
WRITE_SKIP_COUNT BIGINT ,
ROLLBACK_COUNT BIGINT ,

4
spring-batch-core/src/main/resources/schema-hsqldb.sql
View File

@@ -54,7 +54,9 @@ CREATE TABLE BATCH_STEP_EXECUTION (
END_TIME TIMESTAMP DEFAULT NULL ,
STATUS VARCHAR(10) ,
COMMIT_COUNT BIGINT ,
ITEM_COUNT BIGINT ,
READ_COUNT BIGINT ,
FILTER_COUNT BIGINT ,
WRITE_COUNT BIGINT ,
READ_SKIP_COUNT BIGINT ,
WRITE_SKIP_COUNT BIGINT ,
ROLLBACK_COUNT BIGINT ,

4
spring-batch-core/src/main/resources/schema-mysql.sql
View File

@@ -54,7 +54,9 @@ CREATE TABLE BATCH_STEP_EXECUTION (
END_TIME DATETIME DEFAULT NULL ,
STATUS VARCHAR(10) ,
COMMIT_COUNT BIGINT ,
ITEM_COUNT BIGINT ,
READ_COUNT BIGINT ,
FILTER_COUNT BIGINT ,
WRITE_COUNT BIGINT ,
READ_SKIP_COUNT BIGINT ,
WRITE_SKIP_COUNT BIGINT ,
ROLLBACK_COUNT BIGINT ,

4
spring-batch-core/src/main/resources/schema-oracle10g.sql
View File

@@ -54,7 +54,9 @@ CREATE TABLE BATCH_STEP_EXECUTION (
END_TIME TIMESTAMP DEFAULT NULL ,
STATUS VARCHAR2(10) ,
COMMIT_COUNT NUMBER(38) ,
ITEM_COUNT NUMBER(38) ,
READ_COUNT NUMBER(38) ,
FILTER_COUNT NUMBER(38) ,
WRITE_COUNT NUMBER(38) ,
READ_SKIP_COUNT NUMBER(38) ,
WRITE_SKIP_COUNT NUMBER(38) ,
ROLLBACK_COUNT NUMBER(38) ,

4
spring-batch-core/src/main/resources/schema-postgresql.sql
View File

@@ -54,7 +54,9 @@ CREATE TABLE BATCH_STEP_EXECUTION (
END_TIME TIMESTAMP DEFAULT NULL ,
STATUS VARCHAR(10) ,
COMMIT_COUNT BIGINT ,
ITEM_COUNT BIGINT ,
READ_COUNT BIGINT ,
FILTER_COUNT BIGINT ,
WRITE_COUNT BIGINT ,
READ_SKIP_COUNT BIGINT ,
WRITE_SKIP_COUNT BIGINT ,
ROLLBACK_COUNT BIGINT ,

4
spring-batch-core/src/main/resources/schema-sqlserver.sql
View File

@@ -54,7 +54,9 @@ CREATE TABLE BATCH_STEP_EXECUTION (
END_TIME DATETIME DEFAULT NULL ,
STATUS VARCHAR(10) ,
COMMIT_COUNT BIGINT ,
ITEM_COUNT BIGINT ,
READ_COUNT BIGINT ,
FILTER_COUNT BIGINT ,
WRITE_COUNT BIGINT ,
READ_SKIP_COUNT BIGINT ,
WRITE_SKIP_COUNT BIGINT ,
ROLLBACK_COUNT BIGINT ,

4
spring-batch-core/src/main/resources/schema-sybase.sql
View File

@@ -54,7 +54,9 @@ CREATE TABLE BATCH_STEP_EXECUTION (
END_TIME DATETIME DEFAULT NULL NULL,
STATUS VARCHAR(10) NULL,
COMMIT_COUNT BIGINT NULL,
ITEM_COUNT BIGINT NULL,
READ_COUNT BIGINT NULL,
FILTER_COUNT BIGINT NULL,
WRITE_COUNT BIGINT NULL,
READ_SKIP_COUNT BIGINT NULL,
WRITE_SKIP_COUNT BIGINT NULL,
ROLLBACK_COUNT BIGINT NULL,

4
spring-batch-core/src/main/sql/tables-main.sql.vpp
View File

@@ -42,7 +42,9 @@ CREATE TABLE BATCH_STEP_EXECUTION (
END_TIME ${TIMESTAMP} DEFAULT NULL $!{NULL},
STATUS ${VARCHAR}(10) $!{NULL},
COMMIT_COUNT ${BIGINT} $!{NULL},
ITEM_COUNT ${BIGINT} $!{NULL},
READ_COUNT ${BIGINT} $!{NULL},
FILTER_COUNT ${BIGINT} $!{NULL},
WRITE_COUNT ${BIGINT} $!{NULL},
READ_SKIP_COUNT ${BIGINT} $!{NULL},
WRITE_SKIP_COUNT ${BIGINT} $!{NULL},
ROLLBACK_COUNT ${BIGINT} $!{NULL},

6
spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/AbstractStepExecutionDaoTests.java
View File

@@ -99,6 +99,9 @@ public abstract class AbstractStepExecutionDaoTests extends AbstractTransactiona
stepExecution.setWriteSkipCount(5);
stepExecution.setRollbackCount(3);
stepExecution.setLastUpdated(new Date(System.currentTimeMillis()));
stepExecution.setReadCount(17);
stepExecution.setFilterCount(15);
stepExecution.setWriteCount(13);
dao.saveStepExecution(stepExecution);
StepExecution retrieved = dao.getStepExecution(jobExecution, step.getName());
@@ -108,6 +111,9 @@ public abstract class AbstractStepExecutionDaoTests extends AbstractTransactiona
assertEquals(stepExecution.getWriteSkipCount(), retrieved.getWriteSkipCount());
assertEquals(stepExecution.getRollbackCount(), retrieved.getRollbackCount());
assertEquals(stepExecution.getLastUpdated(), retrieved.getLastUpdated());
assertEquals(stepExecution.getReadCount(), retrieved.getReadCount());
assertEquals(stepExecution.getFilterCount(), retrieved.getFilterCount());
assertEquals(stepExecution.getWriteCount(), retrieved.getWriteCount());
assertNull(dao.getStepExecution(jobExecution, "not-existing step"));
}