@Override
	public void afterPropertiesSet() throws Exception {
		if (!StringUtils.hasText(sql)) {
			Assert.hasText(tableName, "tableName must be set");
			Assert.hasText(columnNames, "columns must be set");

			String sql;
			if (StringUtils.hasText(partitionClause)) {
				sql = "SELECT " + columnNames + " FROM " + tableName + " " + partitionClause;
			}
			else {
				sql = "SELECT " + columnNames + " FROM " + tableName;
			}
			log.info("Setting SQL to: " + sql);
			setSql(sql);
		}
		else if (StringUtils.hasText(columnNames) || StringUtils.hasText(tableName)) {
			log.warn("You must set either the 'sql' property or 'tableName' and 'columns'.");
		}

		DatabaseType type = DatabaseType.fromMetaData(dataSource);

		switch (type) {
			case MYSQL:
				fetchSize = Integer.MIN_VALUE;
				// MySql doesn't support getRow for a streaming cursor
				verifyCursorPosition = false;
				break;
			case SQLITE:
				fetchSize = AbstractCursorItemReader.VALUE_NOT_SET;
				break;
			default:
				// keep configured fetchSize
		}

		reader = new NamedColumnJdbcItemReader();
		reader.setSql(sql);
		reader.setFetchSize(fetchSize);
		reader.setDataSource(dataSource);
		reader.setVerifyCursorPosition(verifyCursorPosition);
		reader.setDelimiter(delimiter);
		reader.afterPropertiesSet();

		initialized = true;
	}
Factory for configuring a {@link org.springframework.xd.jdbc.NamedColumnJdbcItemReader}.  This factory
will verify the type of database and configure the appropriate properties so that a streaming cursor is
returned.  Specifically, it will configure {@code fetchSize=Integer.MIN_VALUE} for MySql and set
{@code verifyCursorPosition=false } for both MySql and SQLite.

@author Michael Minella
@author Thomas Risberg
extensions/spring-xd-extension-jdbc/src/main/java/org/springframework/xd/jdbc/NamedColumnJdbcItemReaderFactory.java