[jdbc] Add mssql support for scan operation

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

@@ -1,5 +1,5 @@
 /**
- * Copyright (c) 2010 - 2016 Yahoo! Inc., 2016 YCSB contributors. All rights reserved.
+ * Copyright (c) 2010 - 2016 Yahoo! Inc., 2016, 2019 YCSB contributors. All rights reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License"); you
  * may not use this file except in compliance with the License. You
@@ -82,6 +82,7 @@ public class JdbcDBClient extends DB {
   /** The field name prefix in the table. */
   public static final String COLUMN_PREFIX = "FIELD";
 
+  private boolean sqlserver = false;
   private List<Connection> conns;
   private boolean initialized = false;
   private Properties props;
@@ -183,6 +184,10 @@ public void init() throws DBException {
     String passwd = props.getProperty(CONNECTION_PASSWD, DEFAULT_PROP);
     String driver = props.getProperty(DRIVER_CLASS);
 
+    if (driver.contains("sqlserver")) {
+      sqlserver = true;
+    }
+
     this.jdbcFetchSize = getIntProperty(props, JDBC_FETCH_SIZE);
     this.batchSize = getIntProperty(props, DB_BATCH_SIZE);
 
@@ -299,7 +304,7 @@ private PreparedStatement createAndCacheUpdateStatement(StatementType updateType
 
   private PreparedStatement createAndCacheScanStatement(StatementType scanType, String key)
       throws SQLException {
-    String select = dbFlavor.createScanStatement(scanType, key);
+    String select = dbFlavor.createScanStatement(scanType, key, sqlserver);
     PreparedStatement scanStatement = getShardConnectionByKey(key).prepareStatement(select);
     if (this.jdbcFetchSize > 0) {
       scanStatement.setFetchSize(this.jdbcFetchSize);
@@ -348,8 +353,13 @@ public Status scan(String tableName, String startKey, int recordcount, Set<Strin
       if (scanStatement == null) {
         scanStatement = createAndCacheScanStatement(type, startKey);
       }
-      scanStatement.setString(1, startKey);
-      scanStatement.setInt(2, recordcount);
+      if (sqlserver) {
+        scanStatement.setInt(1, recordcount);
+        scanStatement.setString(2, startKey);
+      } else {
+        scanStatement.setString(1, startKey);
+        scanStatement.setInt(2, recordcount);
+      }
       ResultSet resultSet = scanStatement.executeQuery();
       for (int i = 0; i < recordcount && resultSet.next(); i++) {
         if (result != null && fields != null) {

jdbc/src/main/java/com/yahoo/ycsb/db/flavors/DBFlavor.java

@@ -1,5 +1,5 @@
 /**
- * Copyright (c) 2016 YCSB contributors. All rights reserved.
+ * Copyright (c) 2016, 2019 YCSB contributors. All rights reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License"); you
  * may not use this file except in compliance with the License. You
@@ -65,5 +65,5 @@ public static DBFlavor fromJdbcUrl(String url) {
   /**
    * Create and return a SQL statement for scanning data.
    */
-  public abstract String createScanStatement(StatementType scanType, String key);
+  public abstract String createScanStatement(StatementType scanType, String key, boolean sqlserver);
 }

jdbc/src/main/java/com/yahoo/ycsb/db/flavors/DefaultDBFlavor.java

@@ -1,5 +1,5 @@
 /**
- * Copyright (c) 2016 YCSB contributors. All rights reserved.
+ * Copyright (c) 2016, 2019 YCSB contributors. All rights reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License"); you
  * may not use this file except in compliance with the License. You
@@ -84,15 +84,22 @@ public String createUpdateStatement(StatementType updateType, String key) {
   }
 
   @Override
-  public String createScanStatement(StatementType scanType, String key) {
-    StringBuilder select = new StringBuilder("SELECT * FROM ");
+  public String createScanStatement(StatementType scanType, String key, boolean sqlserver) {
+    StringBuilder select;
+    if (sqlserver) {
+      select = new StringBuilder("SELECT TOP (?) * FROM ");
+    } else {
+      select = new StringBuilder("SELECT * FROM ");
+    }
     select.append(scanType.getTableName());
     select.append(" WHERE ");
     select.append(JdbcDBClient.PRIMARY_KEY);
     select.append(" >= ?");
     select.append(" ORDER BY ");
     select.append(JdbcDBClient.PRIMARY_KEY);
-    select.append(" LIMIT ?");
+    if (!sqlserver) {
+      select.append(" LIMIT ?");
+    }
     return select.toString();
   }
 }