Support running one procedure at a time

src/com/oltpbenchmark/CommandLineOptions.java

@@ -32,6 +32,11 @@ public class CommandLineOptions {
                 "execute",
                 true,
                 "Execute the benchmark workload");
+        COMMAND_LINE_OPTS.addOption(
+                null,
+                "run_one",
+                true,
+                "Run one instance of the process specified by --proc");
 
 
         COMMAND_LINE_OPTS.addOption(
@@ -59,6 +64,11 @@ public class CommandLineOptions {
         COMMAND_LINE_OPTS.addOption(null, "num-connections", true, "Number of connections used");
         COMMAND_LINE_OPTS.addOption(null, "merge-results", true, "Merge results from various output files");
         COMMAND_LINE_OPTS.addOption(null, "dir", true, "Directory containing the csv files");
+
+        COMMAND_LINE_OPTS.addOption(null, "proc", true, "Process to run one of");
+
+        COMMAND_LINE_OPTS.addOption(null, "run_wid", true, "Warehouse ID for single run");
+        COMMAND_LINE_OPTS.addOption(null, "run_did", true, "District ID for single run");
     }
 
     public CommandLineOptions() {}
@@ -69,7 +79,8 @@ public enum Mode {
         CLEAR,
         LOAD,
         EXECUTE,
-        MERGE_RESULTS
+        MERGE_RESULTS,
+        RUN_ONE
     }
 
     void init(String[] args) throws ParseException {
@@ -91,6 +102,7 @@ public Mode getMode() {
         if (isBooleanOptionSet("create")) return Mode.CREATE;
         if (isBooleanOptionSet("load")) return Mode.LOAD;
         if (isBooleanOptionSet("execute")) return Mode.EXECUTE;
+        if (isBooleanOptionSet("run_one")) return Mode.RUN_ONE;
         assert isBooleanOptionSet("merge-results");
         return Mode.MERGE_RESULTS;
     }
@@ -172,4 +184,16 @@ public Optional<String> getDirPath() {
     public boolean getIsOutputMetricHistogramsSet() {
         return isBooleanOptionSet("histograms");
     }
+
+    public Optional<String> getRunOneProc() {
+        return getStringOpt("proc");
+    }
+
+    public Optional<Integer> getRunOneWarehouse() {
+        return getIntOpt("run_wid");
+    }
+
+    public Optional<Integer> getRunDistrict() {
+        return getIntOpt("run_did");
+    }
 }

src/com/oltpbenchmark/DBWorkload.java

@@ -18,17 +18,17 @@
 package com.oltpbenchmark;
 
 import java.io.*;
+import java.lang.reflect.InvocationTargetException;
+import java.sql.Connection;
+import java.sql.SQLException;
 import java.text.DecimalFormat;
 import java.util.*;
 
+import com.oltpbenchmark.api.*;
 import org.apache.commons.collections15.map.ListOrderedMap;
 import org.apache.commons.lang.StringUtils;
 import org.apache.log4j.Logger;
 
-import com.oltpbenchmark.api.BenchmarkModule;
-import com.oltpbenchmark.api.TransactionType;
-import com.oltpbenchmark.api.TransactionTypes;
-import com.oltpbenchmark.api.Worker;
 import com.oltpbenchmark.benchmarks.tpcc.procedures.*;
 import com.oltpbenchmark.jdbc.InstrumentedPreparedStatement;
 import com.oltpbenchmark.util.FileUtil;
@@ -367,6 +367,15 @@ public static void main(String[] args) throws Exception {
       }
     }
 
+    if (options.getMode() == CommandLineOptions.Mode.RUN_ONE) {
+      String procName = options.getRunOneProc().orElseThrow(() -> new RuntimeException("Must specify --proc"));
+      Random r = new Random();
+      int wid = options.getRunOneWarehouse().orElse(r.nextInt(numWarehouses) + 1);
+      int lowerDistrictId = options.getRunDistrict().orElse(1);
+      int upperDistrictId = options.getRunDistrict().orElse(10);
+      runOne(benchList.get(0), procName, wid, numWarehouses, lowerDistrictId, upperDistrictId);
+    }
+
     // Execute Workload
     if (options.getMode() == CommandLineOptions.Mode.EXECUTE) {
       // Bombs away!
@@ -486,6 +495,25 @@ private static Results runWorkload(List<BenchmarkModule> benchList, int interval
     return r;
   }
 
+  private static void runOne(
+        BenchmarkModule module, String txnType, int warehouseID, int numWarehouses, int lowerDistrictId,
+        int upperDistrictId) throws Exception {
+    InstrumentedPreparedStatement.trackLatencyMetrics(true);
+    module.createDataSource();
+    Worker w = new Worker(module, 1, warehouseID, lowerDistrictId, upperDistrictId);
+    Connection conn = module.makeConnection();
+    TransactionType txn = module.initTransactionType(txnType, 1);
+    // TODO -- make sure that connection establishment and prepared statement creation happens, and then wait some time,
+    // and then execute the procedure. This will allow timing info to be more accurate and allow easier tracking of
+    // traces in logs
+    assert txn != null;
+    long startTime = System.nanoTime();
+    w.executeWork(conn, txn);
+    LOG.info("Latency: - " + (System.nanoTime() - startTime) * 0.0000010 );
+    conn.close();
+    txn.getProcedureClass().getMethod("printLatencyStats").invoke(null);
+  }
+
   private static void PrintToplineResults(List<Worker> workers, Results r) {
     long numNewOrderTransactions = 0;
     for (Worker w : workers) {

src/com/oltpbenchmark/api/Procedure.java

@@ -101,8 +101,12 @@ public abstract void run(Connection conn, Random gen,
                              int terminalDistrictLowerID, int terminalDistrictUpperID,
                              Worker w) throws SQLException;
 
+    public abstract void prepareStatements(Connection conn) throws SQLException;
+
     public void test(Connection conn, Worker w) throws Exception {}
 
+    public static void printLatencyStats() {}
+
     /**
      * Thrown from a Procedure to indicate to the Worker
      * that the procedure should be aborted and rolled back.

src/com/oltpbenchmark/api/TransactionType.java

@@ -28,6 +28,9 @@ public void run(
 				Connection conn, Random gen, int terminalWarehouseID, int numWarehouses, int terminalDistrictLowerID,
 				int terminalDistrictUpperID, Worker w) {
 		}
+
+		@Override
+		public void prepareStatements(Connection conn) throws SQLException {}
 	}
     public static final int INVALID_ID = 0;
     public static final TransactionType INVALID = new TransactionType(Invalid.class, INVALID_ID);

src/com/oltpbenchmark/api/Worker.java

@@ -588,7 +588,7 @@ protected final TransactionExecutionState doWork(SubmittedProcedure pieceOfWork)
     /**
      * Executes a single TPCC transaction of type transactionType.
      */
-    protected TransactionStatus executeWork(Connection conn, TransactionType nextTransaction)
+    public TransactionStatus executeWork(Connection conn, TransactionType nextTransaction)
                                             throws UserAbortException, SQLException {
       try {
         Procedure proc = this.getProcedure(nextTransaction.getProcedureClass());

src/com/oltpbenchmark/benchmarks/tpcc/procedures/Delivery.java

@@ -112,13 +112,7 @@ public void run(Connection conn, Random gen,
     boolean trace = LOG.isDebugEnabled();
     int o_carrier_id = TPCCUtil.randomNumber(1, 10, gen);
     Timestamp timestamp = w.getBenchmarkModule().getTimestamp(System.currentTimeMillis());
-    delivGetOrderId = this.getPreparedStatement(conn, delivGetOrderIdSQL);
-    delivDeleteNewOrder =  this.getPreparedStatement(conn, delivDeleteNewOrderSQL);
-    delivGetCustId = this.getPreparedStatement(conn, delivGetCustIdSQL);
-    delivUpdateCarrierId = this.getPreparedStatement(conn, delivUpdateCarrierIdSQL);
-    delivUpdateDeliveryDate = this.getPreparedStatement(conn, delivUpdateDeliveryDateSQL);
-    delivSumOrderAmount = this.getPreparedStatement(conn, delivSumOrderAmountSQL);
-    delivUpdateCustBalDelivCnt = this.getPreparedStatement(conn, delivUpdateCustBalDelivCntSQL);
+    prepareStatements(conn);
 
     int d_id, c_id;
     int[] orderIDs;
@@ -264,4 +258,15 @@ public void run(Connection conn, Random gen,
       LOG.trace(terminalMessage.toString());
     }
   }
+
+  @Override
+  public void prepareStatements(Connection conn) throws SQLException {
+    delivGetOrderId = this.getPreparedStatement(conn, delivGetOrderIdSQL);
+    delivDeleteNewOrder =  this.getPreparedStatement(conn, delivDeleteNewOrderSQL);
+    delivGetCustId = this.getPreparedStatement(conn, delivGetCustIdSQL);
+    delivUpdateCarrierId = this.getPreparedStatement(conn, delivUpdateCarrierIdSQL);
+    delivUpdateDeliveryDate = this.getPreparedStatement(conn, delivUpdateDeliveryDateSQL);
+    delivSumOrderAmount = this.getPreparedStatement(conn, delivSumOrderAmountSQL);
+    delivUpdateCustBalDelivCnt = this.getPreparedStatement(conn, delivUpdateCustBalDelivCntSQL);
+  }
 }

src/com/oltpbenchmark/benchmarks/tpcc/procedures/NewOrder.java

@@ -247,18 +247,22 @@ public void run(Connection conn, Random gen,
     if (TPCCUtil.randomNumber(1, 100, gen) == 1)
       itemIDs[numItems - 1] = TPCCConfig.INVALID_ITEM_ID;
 
-    // Initializing all prepared statements.
+    prepareStatements(conn);
+
+    newOrderTransaction(terminalWarehouseID, districtID,
+                        customerID, numItems, allLocal, itemIDs,
+                        supplierWarehouseIDs, orderQuantities, conn, w);
+  }
+
+  @Override
+  public void prepareStatements(Connection conn) throws SQLException {
     stmtGetCust=this.getPreparedStatement(conn, stmtGetCustSQL);
     stmtGetWhse=this.getPreparedStatement(conn, stmtGetWhseSQL);
     stmtGetDist=this.getPreparedStatement(conn, stmtGetDistSQL);
     stmtInsertNewOrder=this.getPreparedStatement(conn, stmtInsertNewOrderSQL);
     stmtUpdateDist =this.getPreparedStatement(conn, stmtUpdateDistSQL);
     stmtUpdateStock =this.getPreparedStatement(conn, stmtUpdateStockSQL);
     stmtInsertOOrder =this.getPreparedStatement(conn, stmtInsertOOrderSQL);
-
-    newOrderTransaction(terminalWarehouseID, districtID,
-                        customerID, numItems, allLocal, itemIDs,
-                        supplierWarehouseIDs, orderQuantities, conn, w);
   }
 
   private void newOrderTransaction(int w_id, int d_id, int c_id,

src/com/oltpbenchmark/benchmarks/tpcc/procedures/OrderStatus.java

@@ -90,11 +90,7 @@ public void run(Connection conn, Random gen, int w_id, int numWarehouses,
                   Worker w) throws SQLException {
     boolean trace = LOG.isTraceEnabled();
 
-    // initializing all prepared statements
-    payGetCust = this.getPreparedStatement(conn, payGetCustSQL);
-    customerByName = this.getPreparedStatement(conn, customerByNameSQL);
-    ordStatGetNewestOrd = this.getPreparedStatement(conn, ordStatGetNewestOrdSQL);
-    ordStatGetOrderLines = this.getPreparedStatement(conn, ordStatGetOrderLinesSQL);
+    prepareStatements(conn);
 
     int d_id = TPCCUtil.randomNumber(terminalDistrictLowerID, terminalDistrictUpperID, gen);
     boolean c_by_name = false;
@@ -218,6 +214,14 @@ public void run(Connection conn, Random gen, int w_id, int numWarehouses,
 
   }
 
+  @Override
+  public void prepareStatements(Connection conn) throws SQLException {
+    payGetCust = this.getPreparedStatement(conn, payGetCustSQL);
+    customerByName = this.getPreparedStatement(conn, customerByNameSQL);
+    ordStatGetNewestOrd = this.getPreparedStatement(conn, ordStatGetNewestOrdSQL);
+    ordStatGetOrderLines = this.getPreparedStatement(conn, ordStatGetOrderLinesSQL);
+  }
+
   // attention duplicated code across trans... ok for now to maintain separate
   // prepared statements
   public Customer getCustomerById(int c_w_id, int c_d_id, int c_id) throws SQLException {

src/com/oltpbenchmark/benchmarks/tpcc/procedures/Payment.java

@@ -135,10 +135,8 @@ public static void printLatencyStats() {
     LOG.info("latency InsertHist " + payInsertHistSQL.getStats());
     LOG.info("latency CustomerByName " + customerByNameSQL.getStats());
   }
-  public void run(Connection conn, Random gen,
-                  int w_id, int numWarehouses,
-                  int terminalDistrictLowerID, int terminalDistrictUpperID,
-                  Worker w) throws SQLException {
+
+  public void prepareStatements(Connection conn) throws SQLException {
     payUpdateWhse = this.getPreparedStatement(conn, payUpdateWhseSQL);
     payGetWhse = this.getPreparedStatement(conn, payGetWhseSQL);
     payUpdateDist = this.getPreparedStatement(conn, payUpdateDistSQL);
@@ -149,7 +147,13 @@ public void run(Connection conn, Random gen,
     payUpdateCustBal = this.getPreparedStatement(conn, payUpdateCustBalSQL);
     payInsertHist = this.getPreparedStatement(conn, payInsertHistSQL);
     customerByName = this.getPreparedStatement(conn, customerByNameSQL);
+  }
 
+  public void run(Connection conn, Random gen,
+                  int w_id, int numWarehouses,
+                  int terminalDistrictLowerID, int terminalDistrictUpperID,
+                  Worker w) throws SQLException {
+    prepareStatements(conn);
     int districtID = TPCCUtil.randomNumber(terminalDistrictLowerID, terminalDistrictUpperID, gen);
 
     int x = TPCCUtil.randomNumber(1, 100, gen);

src/com/oltpbenchmark/benchmarks/tpcc/procedures/StockLevel.java

@@ -83,10 +83,7 @@ public void run(Connection conn, Random gen,
                   int terminalDistrictLowerID, int terminalDistrictUpperID,
                   Worker w) throws SQLException {
     boolean trace = LOG.isTraceEnabled();
-    stockGetDistOrderId = this.getPreparedStatement(conn, stockGetDistOrderIdSQL);
-    stockGetCountStockFunc = new InstrumentedPreparedStatement(conn.prepareCall(stockGetCountStockSQL.getSqlStmt().getSQL()),
-                                                               stockGetCountStockFuncLatency);
-
+    prepareStatements(conn);
 
     int threshold = TPCCUtil.randomNumber(10, 20, gen);
     int d_id = TPCCUtil.randomNumber(terminalDistrictLowerID,terminalDistrictUpperID, gen);
@@ -140,4 +137,11 @@ public void run(Connection conn, Random gen,
       LOG.trace(terminalMessage);
     }
   }
+
+  @Override
+  public void prepareStatements(Connection conn) throws SQLException {
+    stockGetDistOrderId = this.getPreparedStatement(conn, stockGetDistOrderIdSQL);
+    stockGetCountStockFunc = new InstrumentedPreparedStatement(conn.prepareCall(stockGetCountStockSQL.getSqlStmt().getSQL()),
+            stockGetCountStockFuncLatency);
+  }
 }