- * Simple input source that opens a JDBC cursor and continually retrieves the next row in the ResultSet. It is extremely - * important to note that the JdbcDriver used must be version 3.0 or higher. This is because earlier versions do not - * support holding a ResultSet open over commits. + * Simple input source that opens a JDBC cursor and continually retrieves the + * next row in the ResultSet. It is extremely important to note that the + * JdbcDriver used must be version 3.0 or higher. This is because earlier + * versions do not support holding a ResultSet open over commits. *
* *- * Each call to {@link #read()} will call the provided RowMapper, passing in the ResultSet. There is currently no - * wrapping of the ResultSet to suppress calls to next(). However, if the RowMapper (mistakenly) increments the current - * row, the next call to read will verify that the current row is at the expected position and throw a - * DataAccessException if it is not. This means that, in theory, a RowMapper could read ahead, as long as it returns the - * row back to the correct position before returning. The reason for such strictness on the ResultSet is due to the need - * to maintain control for transactions, restartability and skippability. This ensures that each call to {@link #read()} - * returns the ResultSet at the correct line, regardless of rollbacks, restarts, or skips. + * Each call to {@link #read()} will call the provided RowMapper, passing in the + * ResultSet. There is currently no wrapping of the ResultSet to suppress calls + * to next(). However, if the RowMapper (mistakenly) increments the current row, + * the next call to read will verify that the current row is at the expected + * position and throw a DataAccessException if it is not. This means that, in + * theory, a RowMapper could read ahead, as long as it returns the row back to + * the correct position before returning. The reason for such strictness on the + * ResultSet is due to the need to maintain control for transactions, + * restartability and skippability. This ensures that each call to + * {@link #read()} returns the ResultSet at the correct line, regardless of + * rollbacks, restarts, or skips. *
* *- * {@link ExecutionContext}: The current row is returned as restart data, and when restored from that same data, the - * cursor is opened and the current row set to the value within the restart data. Two values are stored: the current + * {@link ExecutionContext}: The current row is returned as restart data, and + * when restored from that same data, the cursor is opened and the current row + * set to the value within the restart data. Two values are stored: the current * line being processed and the number of lines that have been skipped. *
* *- * Transactions: The same ResultSet is held open regardless of commits or roll backs in a surrounding transaction. This - * means that when such a transaction is committed, the input source is notified through the {@link #mark()} and - * {@link #reset()} so that it can save it's current row number. Later, if the transaction is rolled back, the current - * row can be moved back to the same row number as it was on when commit was called. + * Transactions: The same ResultSet is held open regardless of commits or roll + * backs in a surrounding transaction. This means that when such a transaction + * is committed, the input source is notified through the {@link #mark()} and + * {@link #reset()} so that it can save it's current row number. Later, if the + * transaction is rolled back, the current row can be moved back to the same row + * number as it was on when commit was called. *
* *- * Calling skip will indicate that a record is bad and should not be re-presented to the user if the transaction is - * rolled back. For example, if row 2 is read in, and found to be bad, calling skip will inform the {@link ItemReader}. - * If reading is then continued, and a rollback is necessary because of an error on output, the input source will be - * returned to row 1. Calling read while on row 1 will move the current row to 3, not 2, because 2 has been marked as - * skipped. + * Calling skip will indicate that a record is bad and should not be + * re-presented to the user if the transaction is rolled back. For example, if + * row 2 is read in, and found to be bad, calling skip will inform the + * {@link ItemReader}. If reading is then continued, and a rollback is + * necessary because of an error on output, the input source will be returned to + * row 1. Calling read while on row 1 will move the current row to 3, not 2, + * because 2 has been marked as skipped. *
* *- * Calling close on this {@link ItemStream} will cause all resources it is currently using to be freed. (Connection, - * ResultSet, etc). It is then illegal to call {@link #read()} again until it has been opened. + * Calling close on this {@link ItemStream} will cause all resources it is + * currently using to be freed. (Connection, ResultSet, etc). It is then illegal + * to call {@link #read()} again until it has been opened. + *
+ * + *
+ * Known limitation: when used with Derby
+ * {@link #setVerifyCursorPosition(boolean)} needs to be false
+ * because {@link ResultSet#getRow()} call used for cursor position verification
+ * throws an exception.
*
Creates a default SQLErrorCodeSQLExceptionTranslator for - * the specified DataSource if none is set. + * Return the exception translator for this instance.
Creates a default
+ * SQLErrorCodeSQLExceptionTranslator for the specified DataSource if none
+ * is set.
*/
protected SQLExceptionTranslator getExceptionTranslator() {
if (exceptionTranslator == null) {
if (dataSource != null) {
exceptionTranslator = new SQLErrorCodeSQLExceptionTranslator(dataSource);
- } else {
+ }
+ else {
exceptionTranslator = new SQLStateSQLExceptionTranslator();
}
}
@@ -277,10 +303,11 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
}
/*
- * Throw a SQLWarningException if we're not ignoring warnings, else log the warnings (at debug level).
+ * Throw a SQLWarningException if we're not ignoring warnings, else log the
+ * warnings (at debug level).
*
- * @param warning the warnings object from the current statement. May be null, in which case this
- * method does nothing.
+ * @param warning the warnings object from the current statement. May be
+ * null, in which case this method does nothing.
*
* @see org.springframework.jdbc.SQLWarningException
*/
@@ -289,10 +316,11 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
SQLWarning warningToLog = warnings;
while (warningToLog != null) {
log.debug("SQLWarning ignored: SQL state '" + warningToLog.getSQLState() + "', error code '"
- + warningToLog.getErrorCode() + "', message [" + warningToLog.getMessage() + "]");
+ + warningToLog.getErrorCode() + "', message [" + warningToLog.getMessage() + "]");
warningToLog = warningToLog.getNextWarning();
}
- } else if (warnings != null) {
+ }
+ else if (warnings != null) {
throw new SQLWarningException("Warning not ignored", warnings);
}
}
@@ -320,28 +348,32 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
Assert.notNull(context, "ExecutionContext must not be null");
executeQuery();
initialized = true;
- long processedRowCount = 0;
+ int processedRowCount = 0;
if (context.containsKey(getKey(CURRENT_PROCESSED_ROW))) {
try {
- processedRowCount = context.getLong(getKey(CURRENT_PROCESSED_ROW));
- while(rs.next()){
- if(rs.getRow() == processedRowCount){
+ processedRowCount = Long.valueOf(context.getLong(getKey(CURRENT_PROCESSED_ROW))).intValue();
+ int count = 0;
+ while (rs.next()) {
+ count++;
+ if (count == processedRowCount) {
break;
}
}
- } catch (SQLException se) {
+ }
+ catch (SQLException se) {
throw getExceptionTranslator().translate("Attempted to move ResultSet to last committed row", sql, se);
}
}
-
+
bufferredReader = new BufferredResultSetReader(rs, mapper, processedRowCount);
}
/**
- * Gives the JDBC driver a hint as to the number of rows that should be fetched from the database when more rows are
- * needed for this ResultSet object. If the fetch size specified is zero, the JDBC driver ignores the
- * value.
+ * Gives the JDBC driver a hint as to the number of rows that should be
+ * fetched from the database when more rows are needed for this
+ * ResultSet object. If the fetch size specified is zero, the
+ * JDBC driver ignores the value.
*
* @param fetchSize the number of rows to fetch
* @see ResultSet#setFetchSize(int)
@@ -351,8 +383,8 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
}
/**
- * Sets the limit for the maximum number of rows that any ResultSet object can contain to the given
- * number.
+ * Sets the limit for the maximum number of rows that any
+ * ResultSet object can contain to the given number.
*
* @param maxRows the new max rows limit; zero means there is no limit
* @see Statement#setMaxRows(int)
@@ -362,10 +394,13 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
}
/**
- * Sets the number of seconds the driver will wait for a Statement object to execute to the given
- * number of seconds. If the limit is exceeded, an SQLException is thrown.
+ * Sets the number of seconds the driver will wait for a
+ * Statement object to execute to the given number of
+ * seconds. If the limit is exceeded, an SQLException is
+ * thrown.
*
- * @param queryTimeout seconds the new query timeout limit in seconds; zero means there is no limit
+ * @param queryTimeout seconds the new query timeout limit in seconds; zero
+ * means there is no limit
* @see Statement#setQueryTimeout(int)
*/
public void setQueryTimeout(int queryTimeout) {
@@ -373,7 +408,8 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
}
/**
- * Set whether SQLWarnings should be ignored (only logged) or exception should be thrown.
+ * Set whether SQLWarnings should be ignored (only logged) or exception
+ * should be thrown.
*
* @param ignoreWarnings if TRUE, warnings are ignored
*/
@@ -382,8 +418,8 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
}
/**
- * Allow verification of cursor position after current row is processed by RowMapper or RowCallbackHandler. Default
- * value is TRUE.
+ * Allow verification of cursor position after current row is processed by
+ * RowMapper or RowCallbackHandler. Default value is TRUE.
*
* @param verifyCursorPosition if true, cursor position is verified
*/
@@ -401,23 +437,23 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
}
/**
- * Set the sql statement to be used when creating the cursor. This statement should be a complete and valid Sql
- * statement, as it will be run directly without any modification.
+ * Set the sql statement to be used when creating the cursor. This statement
+ * should be a complete and valid Sql statement, as it will be run directly
+ * without any modification.
*
* @param sql
*/
public void setSql(String sql) {
this.sql = sql;
}
-
+
/**
- * Set the PreparedStatementSetter to use if any parameter values that need to be set in the supplied
- * query.
+ * Set the PreparedStatementSetter to use if any parameter values that need
+ * to be set in the supplied query.
*
* @param preparedStatementSetter
*/
- public void setPreparedStatementSetter(
- PreparedStatementSetter preparedStatementSetter) {
+ public void setPreparedStatementSetter(PreparedStatementSetter preparedStatementSetter) {
this.preparedStatementSetter = preparedStatementSetter;
}
@@ -430,20 +466,22 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
public void setSaveState(boolean saveState) {
this.saveState = saveState;
}
-
-
-
- private class BufferredResultSetReader implements ItemReader{
-
+ private class BufferredResultSetReader implements ItemReader {
+
private ResultSet rs;
+
private RowMapper rowMapper;
+
private List buffer;
+
private int currentIndex;
- private long processedRowCount;
+
+ private int processedRowCount;
+
private int INITIAL_POSITION = -1;
-
- public BufferredResultSetReader(ResultSet rs, RowMapper rowMapper, long processedRowCount) {
+
+ public BufferredResultSetReader(ResultSet rs, RowMapper rowMapper, int processedRowCount) {
Assert.notNull(rs, "The ResultSet must not be null");
Assert.notNull(rowMapper, "The RowMapper must not be null");
this.rs = rs;
@@ -452,34 +490,34 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
currentIndex = INITIAL_POSITION;
this.processedRowCount = processedRowCount;
}
-
- public BufferredResultSetReader(ResultSet rs, RowMapper rowMapper){
+
+ public BufferredResultSetReader(ResultSet rs, RowMapper rowMapper) {
this(rs, rowMapper, 0);
}
- public Object read() throws Exception, UnexpectedInputException,
- NoWorkFoundException, ParseException {
-
+ public Object read() throws Exception, UnexpectedInputException, NoWorkFoundException, ParseException {
+
currentIndex++;
- // if the incremented index reaches out of the buffer, add next item from result set to buffer
- if(buffer.size() == currentIndex){
- try{
- if(!rs.next()){
+ // if the incremented index reaches out of the buffer, add next item
+ // from result set to buffer
+ if (buffer.size() == currentIndex) {
+ try {
+ if (!rs.next()) {
return null;
}
- int currentRow = rs.getRow();
+ int currentRow = processedRowCount + 1;// rs.getRow();
buffer.add(rowMapper.mapRow(rs, currentRow));
verifyCursorPosition(currentRow);
}
- catch(SQLException se){
+ catch (SQLException se) {
throw getExceptionTranslator().translate("Attempt to process next row failed", sql, se);
}
}
-
+
processedRowCount++;
return buffer.get(currentIndex);
}
-
+
public void mark() throws MarkFailedException {
buffer.clear();
currentIndex = INITIAL_POSITION;
@@ -489,8 +527,9 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
processedRowCount -= buffer.size();
currentIndex = INITIAL_POSITION;
}
-
- // Check the result set is in synch with the currentRow attribute. This is
+
+ // Check the result set is in synch with the currentRow attribute. This
+ // is
// important
// to ensure that the user hasn't modified the current row.
private void verifyCursorPosition(long expectedCurrentRow) throws SQLException {
@@ -500,7 +539,7 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
}
}
}
-
+
public long getProcessedRowCount() {
return processedRowCount;
}