gh-47 Fix an issue with JDBC scans

JDBC scans did not instruct the driver to use a specific fetchsize,
resulting in most cases in an OOM; add support for setting fetchsize.

Also add support for disabling autocommit on the JDBC connections,
mostly because Postgres for example doesn't support setting fetchsize
without disabling autocommit.

jdbc/src/main/conf/db.properties

 # Properties file that contains database connection information.
 
 jdbc.driver=org.h2.Driver
+# jdbc.fetchsize=20
 db.url=jdbc:h2:tcp://foo.com:9092/~/h2/ycsb
 db.user=sa
 db.passwd=

jdbc/src/main/java/com/yahoo/ycsb/db/JdbcDBClient.java

@@ -55,6 +55,7 @@ public class JdbcDBClient extends DB implements JdbcDBClientConstants {
   private ArrayList<Connection> conns;
   private boolean initialized = false;
   private Properties props;
+  private Integer jdbcFetchSize;
   private static final String DEFAULT_PROP = "";
   private ConcurrentMap<StatementType, PreparedStatement> cachedStatements;
   
@@ -176,6 +177,19 @@ public class JdbcDBClient extends DB implements JdbcDBClientConstants {
 		String passwd = props.getProperty(CONNECTION_PASSWD, DEFAULT_PROP);
 		String driver = props.getProperty(DRIVER_CLASS);
 
+      String jdbcFetchSizeStr = props.getProperty(JDBC_FETCH_SIZE);
+          if (jdbcFetchSizeStr != null) {
+          try {
+              this.jdbcFetchSize = Integer.parseInt(jdbcFetchSizeStr);
+          } catch (NumberFormatException nfe) {
+              System.err.println("Invalid JDBC fetch size specified: " + jdbcFetchSizeStr);
+              throw new DBException(nfe);
+          }
+      }
+
+      String autoCommitStr = props.getProperty(JDBC_AUTO_COMMIT, Boolean.TRUE.toString());
+      Boolean autoCommit = Boolean.parseBoolean(autoCommitStr);
+
       try {
 		  if (driver != null) {
 	      Class.forName(driver);
@@ -185,9 +199,13 @@ public class JdbcDBClient extends DB implements JdbcDBClientConstants {
           for (String url: urls.split(",")) {
               System.out.println("Adding shard node URL: " + url);
             Connection conn = DriverManager.getConnection(url, user, passwd);
-		    // Since there is no explicit commit method in the DB interface, all
-		    // operations should auto commit.
-		    conn.setAutoCommit(true);
+
+            // Since there is no explicit commit method in the DB interface, all
+            // operations should auto commit, except when explicitly told not to
+            // (this is necessary in cases such as for PostgreSQL when running a
+            // scan workload with fetchSize)
+            conn.setAutoCommit(autoCommit);
+
             shardCount++;
             conns.add(conn);
           }
@@ -289,6 +307,7 @@ public class JdbcDBClient extends DB implements JdbcDBClientConstants {
     select.append(" >= ");
     select.append("?;");
     PreparedStatement scanStatement = getShardConnectionByKey(key).prepareStatement(select.toString());
+    if (this.jdbcFetchSize != null) scanStatement.setFetchSize(this.jdbcFetchSize);
     PreparedStatement stmt = cachedStatements.putIfAbsent(scanType, scanStatement);
     if (stmt == null) return scanStatement;
     else return stmt;

jdbc/src/main/java/com/yahoo/ycsb/db/JdbcDBClientConstants.java

@@ -35,7 +35,13 @@ public interface JdbcDBClientConstants {
   
   /** The password to use for establishing the connection. */
   public static final String CONNECTION_PASSWD = "db.passwd";
-  
+
+  /** The JDBC fetch size hinted to the driver. */
+  public static final String JDBC_FETCH_SIZE = "jdbc.fetchsize";
+
+  /** The JDBC connection auto-commit property for the driver. */
+  public static final String JDBC_AUTO_COMMIT = "jdbc.autocommit";
+
   /** The name of the property for the number of fields in a record. */
   public static final String FIELD_COUNT_PROPERTY="fieldcount";