if (isDateTimeColumn(checkColumnType)) {
nextIncrementalValue = (nextVal == null) ? null
: manager.datetimeToQueryString(nextVal.toString(),
checkColumnType);
} else if (manager.isCharColumn(checkColumnType)) {
throw new ImportException("Character column "
+ "(" + options.getIncrementalTestColumn() + ") can not be used "
+ "to determine which rows to incrementally import.");
} else {
nextIncrementalValue = (nextVal == null) ? null : nextVal.toString();
}
} catch (SQLException sqlE) {
throw new IOException(sqlE);
}
break;
case DateLastModified:
checkColumnType = manager.getColumnTypes(options.getTableName(),
options.getSqlQuery()).get(options.getIncrementalTestColumn());
nextVal = manager.getCurrentDbTimestamp();
if (null == nextVal) {
throw new IOException("Could not get current time from database");
}
nextIncrementalValue = manager.datetimeToQueryString(nextVal.toString(),
checkColumnType);
break;
default:
throw new ImportException("Undefined incremental import type: "
+ incrementalMode);
}
// Build the WHERE clause components that are used to import
// only this incremental section.
StringBuilder sb = new StringBuilder();
String prevEndpoint = options.getIncrementalLastValue();
if (isDateTimeColumn(checkColumnType) && null != prevEndpoint
&& !prevEndpoint.startsWith("\'") && !prevEndpoint.endsWith("\'")) {
// Incremental imports based on date/time should be 'quoted' in
// ANSI SQL. If the user didn't specify single-quotes, put them
// around, here.
prevEndpoint = manager.datetimeToQueryString(prevEndpoint,
checkColumnType);
}
String checkColName = manager.escapeColName(
options.getIncrementalTestColumn());
LOG.info("Incremental import based on column " + checkColName);
if (null != prevEndpoint) {
if (prevEndpoint.equals(nextIncrementalValue)) {
LOG.info("No new rows detected since last import.");
return false;
}
LOG.info("Lower bound value: " + prevEndpoint);
sb.append(checkColName);
switch (incrementalMode) {
case AppendRows:
sb.append(" > ");
break;
case DateLastModified:
sb.append(" >= ");
break;
default:
throw new ImportException("Undefined comparison");
}
sb.append(prevEndpoint);
sb.append(" AND ");
}
if (null != nextIncrementalValue) {
sb.append(checkColName);
switch (incrementalMode) {
case AppendRows:
sb.append(" <= ");
break;
case DateLastModified:
sb.append(" < ");
break;
default:
throw new ImportException("Undefined comparison");
}
sb.append(nextIncrementalValue);
} else {
sb.append(checkColName);
sb.append(" IS NULL ");