/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to you under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.calcite.test.concurrent;

import org.apache.calcite.jdbc.SqlTimeoutException;
import org.apache.calcite.util.Unsafe;
import org.apache.calcite.util.Util;

import org.slf4j.Logger;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.PrintWriter;
import java.io.Reader;
import java.io.StringReader;
import java.io.StringWriter;
import java.io.Writer;
import java.lang.reflect.Constructor;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Timestamp;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Deque;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Properties;
import java.util.StringTokenizer;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * ConcurrentTestCommandScript creates instances of
 * {@link ConcurrentTestCommand} that perform specific actions in a specific
 * order and within the context of a test thread
 * ({@link ConcurrentTestCommandExecutor}).
 *
 * <p>Actions are loaded from a script (see package javadoc for script format).
 *
 * <p>A single ConcurrentTestCommandScript creates commands
 * for multiple threads. Each thread is represented by an integer "thread ID"
 * and, optionally, a String thread name. Thread IDs may take on any positive
 * integer value and may be a sparse set (e.g. 1, 2, 5). Thread names may be any
 * String.
 *
 * <p>When each command is created, it is associated with a thread and given an
 * execution order. Execution order values are positive integers, must be unique
 * within a thread, and may be a sparse set.
 * See {@link ConcurrentTestCommandGenerator#synchronizeCommandSets} for other
 * considerations.
 */
public class ConcurrentTestCommandScript
    extends ConcurrentTestCommandGenerator {
  //~ Static fields/initializers ---------------------------------------------

  private static final String PRE_SETUP_STATE = "pre-setup";
  private static final String SETUP_STATE = "setup";
  private static final String POST_SETUP_STATE = "post-setup";
  private static final String CLEANUP_STATE = "cleanup";
  private static final String POST_CLEANUP_STATE = "post-cleanup";
  private static final String THREAD_STATE = "thread";
  private static final String REPEAT_STATE = "repeat";
  private static final String SQL_STATE = "sql";
  private static final String POST_THREAD_STATE = "post-thread";
  private static final String EOF_STATE = "eof";

  private static final String VAR = "@var";
  private static final String LOCKSTEP = "@lockstep";
  private static final String NOLOCKSTEP = "@nolockstep";
  private static final String ENABLED = "@enabled";
  private static final String DISABLED = "@disabled";
  private static final String SETUP = "@setup";
  private static final String CLEANUP = "@cleanup";
  private static final String END = "@end";
  private static final String THREAD = "@thread";
  private static final String REPEAT = "@repeat";
  private static final String SYNC = "@sync";
  private static final String TIMEOUT = "@timeout";
  private static final String ROWLIMIT = "@rowlimit";
  private static final String PREPARE = "@prepare";
  private static final String PRINT = "@print";
  private static final String FETCH = "@fetch";
  private static final String CLOSE = "@close";
  private static final String SLEEP = "@sleep";
  private static final String ERR = "@err";
  private static final String ECHO = "@echo";
  private static final String INCLUDE = "@include";
  private static final String SHELL = "@shell";
  private static final String PLUGIN = "@plugin";

  private static final String SQL = "";
  private static final String EOF = null;

  private static final StateAction[] STATE_TABLE = {
      new StateAction(
          PRE_SETUP_STATE,
          new StateDatum[]{
              new StateDatum(VAR, PRE_SETUP_STATE),
              new StateDatum(LOCKSTEP, PRE_SETUP_STATE),
              new StateDatum(NOLOCKSTEP, PRE_SETUP_STATE),
              new StateDatum(ENABLED, PRE_SETUP_STATE),
              new StateDatum(DISABLED, PRE_SETUP_STATE),
              new StateDatum(PLUGIN, PRE_SETUP_STATE),
              new StateDatum(SETUP, SETUP_STATE),
              new StateDatum(CLEANUP, CLEANUP_STATE),
              new StateDatum(THREAD, THREAD_STATE)
          }),

      new StateAction(
          SETUP_STATE,
          new StateDatum[]{
              new StateDatum(END, POST_SETUP_STATE),
              new StateDatum(SQL, SETUP_STATE),
              new StateDatum(INCLUDE, SETUP_STATE),
          }),

      new StateAction(
          POST_SETUP_STATE,
          new StateDatum[]{
              new StateDatum(CLEANUP, CLEANUP_STATE),
              new StateDatum(THREAD, THREAD_STATE)
          }),

      new StateAction(
          CLEANUP_STATE,
          new StateDatum[]{
              new StateDatum(END, POST_CLEANUP_STATE),
              new StateDatum(SQL, CLEANUP_STATE),
              new StateDatum(INCLUDE, CLEANUP_STATE),
          }),

      new StateAction(
          POST_CLEANUP_STATE,
          new StateDatum[]{
              new StateDatum(THREAD, THREAD_STATE)
          }),

      new StateAction(
          THREAD_STATE,
          new StateDatum[]{
              new StateDatum(REPEAT, REPEAT_STATE),
              new StateDatum(SYNC, THREAD_STATE),
              new StateDatum(TIMEOUT, THREAD_STATE),
              new StateDatum(ROWLIMIT, THREAD_STATE),
              new StateDatum(PREPARE, THREAD_STATE),
              new StateDatum(PRINT, THREAD_STATE),
              new StateDatum(FETCH, THREAD_STATE),
              new StateDatum(CLOSE, THREAD_STATE),
              new StateDatum(SLEEP, THREAD_STATE),
              new StateDatum(SQL, THREAD_STATE),
              new StateDatum(ECHO, THREAD_STATE),
              new StateDatum(ERR, THREAD_STATE),
              new StateDatum(SHELL, THREAD_STATE),
              new StateDatum(END, POST_THREAD_STATE)
          }),

      new StateAction(
          REPEAT_STATE,
          new StateDatum[]{
              new StateDatum(SYNC, REPEAT_STATE),
              new StateDatum(TIMEOUT, REPEAT_STATE),
              new StateDatum(ROWLIMIT, REPEAT_STATE),
              new StateDatum(PREPARE, REPEAT_STATE),
              new StateDatum(PRINT, REPEAT_STATE),
              new StateDatum(FETCH, REPEAT_STATE),
              new StateDatum(CLOSE, REPEAT_STATE),
              new StateDatum(SLEEP, REPEAT_STATE),
              new StateDatum(SQL, REPEAT_STATE),
              new StateDatum(ECHO, REPEAT_STATE),
              new StateDatum(ERR, REPEAT_STATE),
              new StateDatum(SHELL, REPEAT_STATE),
              new StateDatum(END, THREAD_STATE)
          }),

      new StateAction(
          POST_THREAD_STATE,
          new StateDatum[]{
              new StateDatum(THREAD, THREAD_STATE),
              new StateDatum(EOF, EOF_STATE)
          })
  };

  private static final int FETCH_LEN = FETCH.length();
  private static final int PREPARE_LEN = PREPARE.length();
  private static final int PRINT_LEN = PRINT.length();
  private static final int REPEAT_LEN = REPEAT.length();
  private static final int SLEEP_LEN = SLEEP.length();
  private static final int THREAD_LEN = THREAD.length();
  private static final int TIMEOUT_LEN = TIMEOUT.length();
  private static final int ROWLIMIT_LEN = ROWLIMIT.length();
  private static final int ERR_LEN = ERR.length();
  private static final int ECHO_LEN = ECHO.length();
  private static final int SHELL_LEN = SHELL.length();
  private static final int PLUGIN_LEN = PLUGIN.length();
  private static final int INCLUDE_LEN = INCLUDE.length();
  private static final int VAR_LEN = VAR.length();

  private static final int BUF_SIZE = 1024;
  private static final int REPEAT_READ_AHEAD_LIMIT = 65536;

  private static final char[] SPACES = fill(new char[BUF_SIZE], ' ');
  private static final char[] DASHES = fill(new char[BUF_SIZE], '-');

  // Special "thread ids" for setup & cleanup sections; actually setup &
  // cleanup SQL is executed by the main thread, and neither are in the the
  // thread map.
  private static final Integer SETUP_THREAD_ID = -1;
  private static final Integer CLEANUP_THREAD_ID = -2;

  //~ Instance fields (representing a single script):

  private boolean quiet = false;
  private boolean verbose = false;
  private Boolean lockstep;
  private Boolean disabled;
  private VariableTable vars = new VariableTable();
  private File scriptDirectory;
  private long scriptStartTime = 0;

  private final List<ConcurrentTestPlugin> plugins = new ArrayList<>();
  private final Map<String, ConcurrentTestPlugin> pluginForCommand =
      new HashMap<>();
  private final Map<String, ConcurrentTestPlugin> preSetupPluginForCommand =
      new HashMap<>();
  private final List<String> setupCommands = new ArrayList<>();
  private final List<String> cleanupCommands = new ArrayList<>();

  private final Map<Integer, BufferedWriter> threadBufferedWriters =
      new HashMap<>();
  private final Map<Integer, StringWriter> threadStringWriters = new HashMap<>();
  private final Map<Integer, ResultsReader> threadResultsReaders =
      new HashMap<>();

  //~ Constructors -----------------------------------------------------------

  public ConcurrentTestCommandScript() throws IOException {
    super();
  }

  /**
   * Constructs and prepares a new ConcurrentTestCommandScript.
   */
  public ConcurrentTestCommandScript(String filename) throws IOException {
    this();
    prepare(filename, null);
  }

  //~ Methods ----------------------------------------------------------------

  private static char[] fill(char[] chars, char c) {
    Arrays.fill(chars, c);
    return chars;
  }

  /**
   * Runs an external application process.
   *
   * @param pb        ProcessBuilder for the application
   * @param logger    if not null, command and exit status will be logged here
   * @param appInput  if not null, data will be copied to application's stdin
   * @param appOutput if not null, data will be captured from application's
   *                  stdout and stderr
   * @return application process exit value
   */
  static int runAppProcess(
      ProcessBuilder pb,
      Logger logger,
      Reader appInput,
      Writer appOutput) throws IOException, InterruptedException {
    pb.redirectErrorStream(true);
    if (logger != null) {
      logger.info("start process: " + pb.command());
    }
    Process p = pb.start();

    // Setup the input/output streams to the subprocess.
    // The buffering here is arbitrary. Javadocs strongly encourage
    // buffering, but the size needed is very dependent on the
    // specific application being run, the size of the input
    // provided by the caller, and the amount of output expected.
    // Since this method is currently used only by unit tests,
    // large-ish fixed buffer sizes have been chosen. If this
    // method becomes used for something in production, it might
    // be better to have the caller provide them as arguments.
    if (appInput != null) {
      OutputStream out =
          new BufferedOutputStream(
              p.getOutputStream(),
              100 * 1024);
      int c;
      while ((c = appInput.read()) != -1) {
        out.write(c);
      }
      out.flush();
    }
    if (appOutput != null) {
      InputStream in =
          new BufferedInputStream(
              p.getInputStream(),
              100 * 1024);
      int c;
      while ((c = in.read()) != -1) {
        appOutput.write(c);
      }
      appOutput.flush();
      in.close();
    }
    p.waitFor();

    int status = p.exitValue();
    if (logger != null) {
      logger.info("exit status=" + status + " from " + pb.command());
    }
    return status;
  }

  /**
   * Gets ready to execute: loads script FILENAME applying external variable
   * BINDINGS
   */
  private void prepare(String filename, List<String> bindings)
      throws IOException {
    vars = new VariableTable();
    CommandParser parser = new CommandParser();
    parser.rememberVariableRebindings(bindings);
    parser.load(filename);

    for (Integer threadId : getThreadIds()) {
      addThreadWriters(threadId);
    }

    // Backwards compatible: printed results always has a setup section, but
    // cleanup section is optional:
    setThreadName(SETUP_THREAD_ID, "setup");
    addThreadWriters(SETUP_THREAD_ID);
    if (!cleanupCommands.isEmpty()) {
      setThreadName(CLEANUP_THREAD_ID, "cleanup");
      addThreadWriters(CLEANUP_THREAD_ID);
    }
  }

  /**
   * Executes the script
   */
  public void execute() throws Exception {
    scriptStartTime = System.currentTimeMillis();
    executeSetup();
    ConcurrentTestCommandExecutor[] threads = innerExecute();
    executeCleanup();
    postExecute(threads);
  }

  private void addThreadWriters(Integer threadId) {
    StringWriter w = new StringWriter();
    BufferedWriter bw = new BufferedWriter(w);
    threadStringWriters.put(threadId, w);
    threadBufferedWriters.put(threadId, bw);
    threadResultsReaders.put(threadId, new ResultsReader(bw));
  }

  public void setQuiet(boolean val) {
    quiet = val;
  }

  public void setVerbose(boolean val) {
    verbose = val;
  }


  public boolean useLockstep() {
    return lockstep != null && lockstep;
  }

  public boolean isDisabled() {
    for (ConcurrentTestPlugin plugin : plugins) {
      if (plugin.isTestDisabled()) {
        return true;
      }
    }

    return disabled != null && disabled;
  }

  public void executeSetup() throws Exception {
    executeCommands(SETUP_THREAD_ID, setupCommands);
  }

  public void executeCleanup() throws Exception {
    executeCommands(CLEANUP_THREAD_ID, cleanupCommands);
  }

  protected void executeCommands(int threadId, List<String> commands)
      throws Exception {
    if (commands == null || commands.size() == 0) {
      return;
    }

    Connection connection = DriverManager.getConnection(jdbcURL, jdbcProps);
    if (connection.getMetaData().supportsTransactions()) {
      connection.setAutoCommit(false);
    }

    boolean forced = false;         // flag, keep going after an error
    try {
      for (String command : commands) {
        String sql = command.trim();
        storeSql(threadId, sql);

        if (isComment(sql)) {
          continue;
        }

        // handle sqlline-type directives:
        if (sql.startsWith("!set")) {
          String[] tokens = sql.split(" +");
          // handle only SET FORCE
          if ((tokens.length > 2)
              && tokens[1].equalsIgnoreCase("force")) {
            forced = asBoolValue(tokens[2]);
          }
          continue;           // else ignore
        } else if (sql.startsWith("!")) {
          continue;           // else ignore
        }

        if (sql.endsWith(";")) {
          sql = sql.substring(0, sql.length() - 1);
        }

        if (isSelect(sql)) {
          try (Statement stmt = connection.createStatement()) {
            ResultSet rset = stmt.executeQuery(sql);
            storeResults(threadId, rset, -1);
          }
        } else if (sql.equalsIgnoreCase("commit")) {
          connection.commit();
        } else if (sql.equalsIgnoreCase("rollback")) {
          connection.rollback();
        } else {
          try (Statement stmt = connection.createStatement()) {
            int rows = stmt.executeUpdate(sql);
            if (rows != 1) {
              storeMessage(
                  threadId, String.valueOf(rows) + " rows affected.");
            } else {
              storeMessage(threadId, "1 row affected.");
            }
          } catch (SQLException ex) {
            if (forced) {
              storeMessage(threadId, ex.getMessage()); // swallow
            } else {
              throw ex;
            }
          }
        }
      }
    } finally {
      if (connection.getMetaData().supportsTransactions()) {
        connection.rollback();
      }
      connection.close();
    }
  }

  // timeout < 0 means no timeout
  private void storeResults(Integer threadId, ResultSet rset, long timeout)
      throws SQLException {
    ResultsReader r = threadResultsReaders.get(threadId);
    r.read(rset, timeout);
  }

  /**
   * Identifies the start of a comment line; same rules as sqlline
   */
  private boolean isComment(String line) {
    return line.startsWith("--") || line.startsWith("#");
  }

  /**
   * translates argument of !set force etc.
   */
  private boolean asBoolValue(String s) {
    return s.equalsIgnoreCase("true")
        || s.equalsIgnoreCase("yes")
        || s.equalsIgnoreCase("on");
  }

  /**
   * Determines whether a block of SQL is a SELECT statement.
   */
  private boolean isSelect(String sql) {
    BufferedReader rdr = new BufferedReader(new StringReader(sql));

    try {
      String line;
      while ((line = rdr.readLine()) != null) {
        line = line.trim().toLowerCase(Locale.ROOT);
        if (isComment(line)) {
          continue;
        }
        return line.startsWith("select")
            || line.startsWith("values")
            || line.startsWith("explain");
      }
    } catch (IOException e) {
      assert false : "IOException via StringReader";
    } finally {
      try {
        rdr.close();
      } catch (IOException e) {
        assert false : "IOException via StringReader";
      }
    }

    return false;
  }

  /**
   * Builds a map of thread ids to result data for the thread. Each result
   * datum is an <code>String[2]</code> containing the thread name and the
   * thread's output.
   *
   * @return the map.
   */
  private Map<Integer, String[]> collectResults() {
    final TreeMap<Integer, String[]> results = new TreeMap<>();

    // get all normal threads
    final TreeSet<Integer> threadIds = new TreeSet<>(getThreadIds());
    // add the "special threads"
    threadIds.add(SETUP_THREAD_ID);
    threadIds.add(CLEANUP_THREAD_ID);

    for (Integer threadId : threadIds) {
      try {
        BufferedWriter bout = threadBufferedWriters.get(threadId);
        if (bout != null) {
          bout.flush();
        }
      } catch (IOException e) {
        assert false : "IOException via StringWriter";
      }
      String threadName = getFormattedThreadName(threadId);
      StringWriter out = threadStringWriters.get(threadId);
      if (out == null) {
        continue;
      }
      results.put(
          threadId,
          new String[]{threadName, out.toString()});
    }
    return results;
  }

  // solely for backwards-compatible output
  private String getFormattedThreadName(Integer id) {
    if (id < 0) {                   // special thread
      return getThreadName(id);
    } else {                        // normal thread
      return "thread " + getThreadName(id);
    }
  }

  public void printResults(PrintWriter out) throws IOException {
    final Map<Integer, String[]> results = collectResults();
    if (verbose) {
      out.write(
          String.format(Locale.ROOT,
              "script execution started at %tc (%d)%n",
              new Timestamp(scriptStartTime), scriptStartTime));
    }
    printThreadResults(out, results.get(SETUP_THREAD_ID));
    for (Integer id : results.keySet()) {
      if (id < 0) {
        continue;               // special thread
      }
      printThreadResults(out, results.get(id)); // normal thread
    }
    printThreadResults(out, results.get(CLEANUP_THREAD_ID));
  }

  private void printThreadResults(PrintWriter out, String[] threadResult)
      throws IOException {
    if (threadResult == null) {
      return;
    }
    String threadName = threadResult[0];
    out.write("-- " + threadName);
    out.println();
    out.write(threadResult[1]);
    out.write("-- end of " + threadName);
    out.println();
    out.println();
    out.flush();
  }

  /**
   * Causes errors to be send here for custom handling. See
   * {@link #customErrorHandler(ConcurrentTestCommandExecutor)}.
   */
  boolean requiresCustomErrorHandling() {
    return true;
  }

  void customErrorHandler(
      ConcurrentTestCommandExecutor executor) {
    StringBuilder message = new StringBuilder();
    Throwable cause = executor.getFailureCause();
    ConcurrentTestCommand command = executor.getFailureCommand();

    if ((command == null) || !command.isFailureExpected()) {
      message.append(cause.getMessage());
      StackTraceElement[] trace = cause.getStackTrace();
      for (StackTraceElement aTrace : trace) {
        message.append("\n\t").append(aTrace.toString());
      }
    } else {
      message.append(cause.getClass().getName())
          .append(": ")
          .append(cause.getMessage());
    }

    storeMessage(
        executor.getThreadId(),
        message.toString());
  }

  /**
   * Retrieves the output stream for the given thread id.
   *
   * @return a BufferedWriter on a StringWriter for the thread.
   */
  private BufferedWriter getThreadWriter(Integer threadId) {
    assert threadBufferedWriters.containsKey(threadId);
    return threadBufferedWriters.get(threadId);
  }


  /**
   * Saves a SQL command to be printed with the thread's output.
   */
  private void storeSql(Integer threadId, String sql) {
    StringBuilder message = new StringBuilder();

    BufferedReader rdr = new BufferedReader(new StringReader(sql));

    try {
      String line;
      while ((line = rdr.readLine()) != null) {
        line = line.trim();

        if (message.length() > 0) {
          message.append('\n');
        }

        message.append("> ").append(line);
      }
    } catch (IOException e) {
      assert false : "IOException via StringReader";
    } finally {
      try {
        rdr.close();
      } catch (IOException e) {
        assert false : "IOException via StringReader";
      }
    }

    storeMessage(
        threadId,
        message.toString());
  }

  /**
   * Saves a message to be printed with the thread's output.
   */
  private void storeMessage(Integer threadId, String message) {
    BufferedWriter out = getThreadWriter(threadId);
    try {
      if (verbose) {
        long t = System.currentTimeMillis() - scriptStartTime;
        out.write("at " + t + ": ");
      }
      out.write(message);
      out.newLine();
    } catch (IOException e) {
      assert false : "IOException on StringWriter";
    }
  }

  //~ Inner Classes ----------------------------------------------------------

  /** State action. */
  private static class StateAction {
    final String state;
    final StateDatum[] stateData;

    StateAction(String state, StateDatum[] stateData) {
      this.state = state;
      this.stateData = stateData;
    }
  }

  /** State datum. */
  private static class StateDatum {
    final String x;
    final String y;

    StateDatum(String x, String y) {
      this.x = x;
      this.y = y;
    }
  }


  /** Symbol table of script variables. */
  private class VariableTable {
    private final Map<String, String> map;

    // matches $$, $var, ${var}
    private final Pattern symbolPattern =
        Pattern.compile("\\$((\\$)|([A-Za-z]\\w*)|\\{([A-Za-z]\\w*)\\})");

    VariableTable() {
      map = new HashMap<>();
    }

    /** Exception. */
    public class Excn extends IllegalArgumentException {
      Excn(String msg) {
        super(msg);
      }
    }

    public boolean isEmpty() {
      return map.isEmpty();
    }

    public boolean isDefined(String sym) {
      return map.containsKey(sym);
    }

    // a symbol must be explicitly defined before it can be set or read.
    public void define(String sym, String val) {
      if (isDefined(sym)) {
        throw new Excn("second declaration of variable " + sym);
      }
      // convert a null val to a null string
      map.put(sym, val == null ? "" : val);
    }

    // returns null is SYM is not defined
    public String get(String sym) {
      if (isDefined(sym)) {
        return map.get(sym);
      } else {
        return null;
      }
    }

    public void set(String sym, String val) {
      if (isDefined(sym)) {
        map.put(sym, val);
        return;
      }
      throw new Excn("undeclared variable " + sym);
    }

    public String expand(String in) {
      if (in.contains("$")) {
        StringBuilder out = new StringBuilder();
        Matcher matcher = symbolPattern.matcher(in);
        int lastEnd = 0;
        while (matcher.find()) {
          int start = matcher.start();
          int end = matcher.end();
          String val;
          if (null != matcher.group(2)) {
            val = "$";          // matched $$
          } else {
            String var = matcher.group(3); // matched $var
            if (var == null) {
              var = matcher.group(4); // matched ${var}
            }
            if (map.containsKey(var)) {
              val = map.get(var);
              val = expand(val);
            } else {
              // not our var, so can't expand
              val = matcher.group(0);
            }
          }
          out.append(in.substring(lastEnd, start));
          out.append(val);
          lastEnd = end;
        }
        out.append(in.substring(lastEnd));
        return out.toString();
      } else {
        return in;
      }
    }
  }


  /** Command parser. */
  private class CommandParser {
    final Pattern splitWords = Pattern.compile("\\s+");
    final Pattern splitBinding = Pattern.compile("=");
    final Pattern matchesVarDefn =
        Pattern.compile("([A-Za-z]\\w*) *=(.*)$");
    // \1 is VAR, \2 is VAL

    // parser state
    private String state;
    private int threadId;
    private int nextThreadId;
    private int order;
    private int repeatCount;
    private boolean scriptHasVars;
    private final Deque<File> currentDirectory = new ArrayDeque<>();

    /** Binding of a value to a variable. */
    private class Binding {
      public final String var;
      public final String val;

      Binding(String var, String val) {
        this.var = var;
        this.val = val;
      }

      // @param phrase has form VAR=VAL
      Binding(String phrase) {
        String[] parts = splitBinding.split(phrase);
        assert parts.length == 2;
        this.var = parts[0];
        this.val = parts[1];
      }
    }

    // A list of Bindings that must be applied immediately after parsing
    // last @var.
    private final List<Binding> deferredBindings = new ArrayList<>();

    CommandParser() {
      state = PRE_SETUP_STATE;
      threadId = nextThreadId = 1;
      order = 1;
      repeatCount = 0;
      scriptHasVars = false;
      currentDirectory.push(null);
    }

    // Parses a set of VAR=VAL pairs from the command line, and saves it for
    // later application.
    public void rememberVariableRebindings(List<String> pairs) {
      if (pairs == null) {
        return;
      }
      for (String pair : pairs) {
        deferredBindings.add(new Binding(pair));
      }
    }

    // to call after all @var commands but before any SQL.
    private void applyVariableRebindings() {
      for (Binding binding : deferredBindings) {
        vars.set(binding.var, binding.val);
      }
    }

    // trace loading of a script
    private void trace(String prefix, Object message) {
      if (verbose && !quiet) {
        if (prefix != null) {
          System.out.print(prefix + ": ");
        }
        System.out.println(message);
      }
    }

    private void trace(String message) {
      trace(null, message);
    }

    /**
     * Parses a multi-threaded script and converts it into test commands.
     */
    private void load(String scriptFileName) throws IOException {
      File scriptFile = new File(currentDirectory.peek(), scriptFileName);
      currentDirectory.push(scriptDirectory = scriptFile.getParentFile());
      try (BufferedReader in = Util.reader(scriptFile)) {
        String line;
        while ((line = in.readLine()) != null) {
          line = line.trim();
          Map<String, String> commandStateMap = lookupState(state);
          final String command;
          boolean isSql = false;
          if (line.equals("") || line.startsWith("--")) {
            continue;
          } else if (line.startsWith("@")) {
            command = firstWord(line);
          } else {
            isSql = true;
            command = SQL;
          }
          if (!commandStateMap.containsKey(command)) {
            throw new IllegalStateException(
                command + " not allowed in state " + state);
          }

          boolean changeState;
          if (isSql) {
            String sql = readSql(line, in);
            loadSql(sql);
            changeState = true;
          } else {
            changeState = loadCommand(command, line, in);
          }
          if (changeState) {
            String nextState = commandStateMap.get(command);
            assert nextState != null;
            if (!nextState.equals(state)) {
              doEndOfState(state);
            }
            state = nextState;
          }
        }

        // at EOF
        currentDirectory.pop();
        if (currentDirectory.size() == 1) {
          // at top EOF
          if (!lookupState(state).containsKey(EOF)) {
            throw new IllegalStateException(
                "Premature end of file in '" + state + "' state");
          }
        }
      }
    }

    private void loadSql(String sql) {
      if (SETUP_STATE.equals(state)) {
        trace("@setup", sql);
        setupCommands.add(sql);
      } else if (CLEANUP_STATE.equals(state)) {
        trace("@cleanup", sql);
        cleanupCommands.add(sql);
      } else if (
          THREAD_STATE.equals(state) || REPEAT_STATE.equals(state)) {
        boolean isSelect = isSelect(sql);
        trace(sql);
        for (int i = threadId; i < nextThreadId; i++) {
          CommandWithTimeout cmd =
              isSelect ? new SelectCommand(sql) : new SqlCommand(sql);
          addCommand(i, order, cmd);
        }
        order++;
      } else {
        assert false;
      }
    }

    // returns TRUE when load-state should advance, FALSE when it must not.
    private boolean loadCommand(
        String command, String line, BufferedReader in) throws IOException {
      if (VAR.equals(command)) {
        String args = line.substring(VAR_LEN).trim();
        scriptHasVars = true;
        trace("@var", args);
        defineVariables(args);

      } else if (LOCKSTEP.equals(command)) {
        assert lockstep == null
            : LOCKSTEP + " and " + NOLOCKSTEP + " may only appear once";
        lockstep = Boolean.TRUE;
        trace("lockstep");

      } else if (NOLOCKSTEP.equals(command)) {
        assert lockstep == null
            : LOCKSTEP + " and " + NOLOCKSTEP + " may only appear once";
        lockstep = Boolean.FALSE;
        trace("no lockstep");

      } else if (DISABLED.equals(command)) {
        assert disabled == null
            : DISABLED + " and " + ENABLED + " may only appear once";
        disabled = Boolean.TRUE;
        trace("disabled");

      } else if (ENABLED.equals(command)) {
        assert disabled == null
            : DISABLED + " and " + ENABLED + " may only appear once";
        disabled = Boolean.FALSE;
        trace("enabled");

      } else if (SETUP.equals(command)) {
        trace("@setup");

      } else if (CLEANUP.equals(command)) {
        trace("@cleanup");

      } else if (INCLUDE.equals(command)) {
        String includedFile =
            vars.expand(line.substring(INCLUDE_LEN).trim());
        trace("@include", includedFile);
        load(includedFile);
        trace("end @include", includedFile);

      } else if (THREAD.equals(command)) {
        String threadNamesStr = line.substring(THREAD_LEN).trim();
        trace("@thread", threadNamesStr);
        StringTokenizer threadNamesTok =
            new StringTokenizer(threadNamesStr, ",");
        while (threadNamesTok.hasMoreTokens()) {
          setThreadName(
              nextThreadId++,
              threadNamesTok.nextToken());
        }

      } else if (REPEAT.equals(command)) {
        String arg = line.substring(REPEAT_LEN).trim();
        repeatCount = Integer.parseInt(vars.expand(arg));
        trace("start @repeat block", repeatCount);
        assert repeatCount > 0 : "Repeat count must be > 0";
        in.mark(REPEAT_READ_AHEAD_LIMIT);

      } else if (END.equals(command)) {
        if (SETUP_STATE.equals(state)) {
          trace("end @setup");
        } else if (CLEANUP_STATE.equals(state)) {
          trace("end @cleanup");
        } else if (THREAD_STATE.equals(state)) {
          threadId = nextThreadId;
        } else if (REPEAT_STATE.equals(state)) {
          trace("repeating");
          repeatCount--;
          if (repeatCount > 0) {
            try {
              in.reset();
            } catch (IOException e) {
              throw new IllegalStateException(
                  "Unable to reset reader -- repeat "
                  + "contents must be less than "
                  + REPEAT_READ_AHEAD_LIMIT + " bytes");
            }

            trace("end @repeat block");
            return false;   // don't change the state
          }
        } else {
          assert false;
        }

      } else if (SYNC.equals(command)) {
        trace("@sync");
        for (int i = threadId; i < nextThreadId; i++) {
          addSynchronizationCommand(i, order);
        }
        order++;

      } else if (TIMEOUT.equals(command)) {
        String args = line.substring(TIMEOUT_LEN).trim();
        String millisStr = vars.expand(firstWord(args));
        long millis = Long.parseLong(millisStr);
        assert millis >= 0L : "Timeout must be >= 0";

        String sql = readSql(skipFirstWord(args).trim(), in);
        trace("@timeout", sql);
        boolean isSelect = isSelect(sql);
        for (int i = threadId; i < nextThreadId; i++) {
          CommandWithTimeout cmd =
              isSelect ? new SelectCommand(sql, millis)
                  : new SqlCommand(sql, millis);
          addCommand(i, order, cmd);
        }
        order++;

      } else if (ROWLIMIT.equals(command)) {
        String args = line.substring(ROWLIMIT_LEN).trim();
        String limitStr = vars.expand(firstWord(args));
        int limit = Integer.parseInt(limitStr);
        assert limit >= 0 : "Rowlimit must be >= 0";

        String sql = readSql(skipFirstWord(args).trim(), in);
        trace("@rowlimit ", sql);
        boolean isSelect = isSelect(sql);
        if (!isSelect) {
          throw new IllegalStateException(
              "Only select can be used with rowlimit");
        }
        for (int i = threadId; i < nextThreadId; i++) {
          addCommand(i, order, new SelectCommand(sql, 0, limit));
        }
        order++;

      } else if (PRINT.equals(command)) {
        String spec = vars.expand(line.substring(PRINT_LEN).trim());
        trace("@print", spec);
        for (int i = threadId; i < nextThreadId; i++) {
          addCommand(i, order, new PrintCommand(spec));
        }
        order++;

      } else if (PREPARE.equals(command)) {
        String startOfSql =
            line.substring(PREPARE_LEN).trim();
        String sql = readSql(startOfSql, in);
        trace("@prepare", sql);
        for (int i = threadId; i < nextThreadId; i++) {
          addCommand(i, order, new PrepareCommand(sql));
        }
        order++;

      } else if (PLUGIN.equals(command)) {
        String cmd = line.substring(PLUGIN_LEN).trim();
        String pluginName = readLine(cmd, in).trim();
        trace("@plugin", pluginName);
        plugin(pluginName);

      } else if (pluginForCommand.containsKey(command)) {
        String cmd = line.substring(command.length())
            .trim();
        cmd = readLine(cmd, in);
        trace("@" + command, cmd);
        for (int i = threadId; i < nextThreadId; i++) {
          addCommand(
              i,
              order,
              new PluginCommand(
                  command, cmd));
        }
        order++;

      } else if (preSetupPluginForCommand.containsKey(command)) {
        String cmd = line.substring(command.length()).trim();
        cmd = readLine(cmd, in);
        trace("@" + command, cmd);
        ConcurrentTestPlugin plugin =
            preSetupPluginForCommand.get(command);
        plugin.preSetupFor(command, cmd);


      } else if (SHELL.equals(command)) {
        String cmd = line.substring(SHELL_LEN).trim();
        cmd = readLine(cmd, in);
        trace("@shell", cmd);
        for (int i = threadId; i < nextThreadId; i++) {
          addCommand(i, order, new ShellCommand(cmd));
        }
        order++;

      } else if (ECHO.equals(command)) {
        String msg = line.substring(ECHO_LEN).trim();
        msg = readLine(msg, in);
        trace("@echo", msg);
        for (int i = threadId; i < nextThreadId; i++) {
          addCommand(i, order, new EchoCommand(msg));
        }
        order++;

      } else if (ERR.equals(command)) {
        String startOfSql =
            line.substring(ERR_LEN).trim();
        String sql = readSql(startOfSql, in);
        trace("@err ", sql);
        boolean isSelect = isSelect(sql);
        for (int i = threadId; i < nextThreadId; i++) {
          CommandWithTimeout cmd =
              isSelect ? new SelectCommand(sql, true)
                  : new SqlCommand(sql, true);
          addCommand(i, order, cmd);
        }
        order++;

      } else if (FETCH.equals(command)) {
        String arg = vars.expand(line.substring(FETCH_LEN).trim());
        trace("@fetch", arg);
        long millis = 0L;
        if (arg.length() > 0) {
          millis = Long.parseLong(arg);
          assert millis >= 0L : "Fetch timeout must be >= 0";
        }

        for (int i = threadId; i < nextThreadId; i++) {
          addCommand(
              i,
              order,
              new FetchAndPrintCommand(millis));
        }
        order++;

      } else if (CLOSE.equals(command)) {
        trace("@close");
        for (int i = threadId; i < nextThreadId; i++) {
          addCloseCommand(i, order);
        }
        order++;

      } else if (SLEEP.equals(command)) {
        String arg = vars.expand(line.substring(SLEEP_LEN).trim());
        trace("@sleep", arg);
        long millis = Long.parseLong(arg);
        assert millis >= 0L : "Sleep timeout must be >= 0";

        for (int i = threadId; i < nextThreadId; i++) {
          addSleepCommand(i, order, millis);
        }
        order++;

      } else {
        assert false : "Unknown command " + command;
      }

      return true;                // normally, advance the state
    }

    private void doEndOfState(String state) {
      if (state.equals(PRE_SETUP_STATE)) {
        applyVariableRebindings();
      }
    }

    private void defineVariables(String line) {
      // two forms: "VAR VAR*" and "VAR=VAL$"
      Matcher varDefn = matchesVarDefn.matcher(line);
      if (varDefn.lookingAt()) {
        String var = varDefn.group(1);
        String val = varDefn.group(2);
        vars.define(var, val);
      } else {
        String[] words = splitWords.split(line);
        for (String var : words) {
          String value = System.getenv(var);
          vars.define(var, value);
        }
      }
    }

    private void plugin(String pluginName) throws IOException {
      try {
        @SuppressWarnings("unchecked")
        Class<ConcurrentTestPlugin> pluginClass =
            (Class<ConcurrentTestPlugin>) Class.forName(pluginName);
        final Constructor<ConcurrentTestPlugin> constructor =
            pluginClass.getConstructor();
        final ConcurrentTestPlugin plugin = constructor.newInstance();
        plugins.add(plugin);
        addExtraCommands(plugin.getSupportedThreadCommands(), THREAD_STATE);
        addExtraCommands(plugin.getSupportedThreadCommands(), REPEAT_STATE);
        for (String commandName : plugin.getSupportedThreadCommands()) {
          pluginForCommand.put(commandName, plugin);
        }
        addExtraCommands(
            plugin.getSupportedPreSetupCommands(), PRE_SETUP_STATE);
        for (String commandName : plugin.getSupportedPreSetupCommands()) {
          preSetupPluginForCommand.put(commandName, plugin);
        }
      } catch (Exception e) {
        throw new IOException(e.toString());
      }
    }

    private void addExtraCommands(Iterable<String> commands, String state) {
      assert state != null;

      for (int i = 0, n = STATE_TABLE.length; i < n; i++) {
        if (state.equals(STATE_TABLE[i].state)) {
          StateDatum[] stateData = STATE_TABLE[i].stateData;
          final List<StateDatum> stateDataList =
              new ArrayList<>(Arrays.asList(stateData));
          for (String cmd : commands) {
            stateDataList.add(new StateDatum(cmd, state));
          }
          STATE_TABLE[i] =
              new StateAction(
                  state, stateDataList.toArray(stateData));
        }
      }
    }

    /**
     * Manages state transitions.
     * Converts a state name into a map. Map keys are the names of available
     * commands (e.g. @sync), and map values are the state to switch to open
     * seeing the command.
     */
    private Map<String, String> lookupState(String state) {
      assert state != null;

      for (StateAction a : STATE_TABLE) {
        if (state.equals(a.state)) {
          StateDatum[] stateData = a.stateData;

          Map<String, String> result = new HashMap<String, String>();
          for (StateDatum datum : stateData) {
            result.put(datum.x, datum.y);
          }
          return result;
        }
      }

      throw new IllegalArgumentException();
    }

    /**
     * Returns the first word of the given line, assuming the line is
     * trimmed. Returns the characters up the first non-whitespace
     * character in the line.
     */
    private String firstWord(String trimmedLine) {
      return trimmedLine.replaceFirst("\\s.*", "");
    }

    /**
     * Returns everything but the first word of the given line, assuming the
     * line is trimmed. Returns the characters following the first series of
     * consecutive whitespace characters in the line.
     */
    private String skipFirstWord(String trimmedLine) {
      return trimmedLine.replaceFirst("^\\S+\\s+", "");
    }

    /**
     * Returns an input line, possible extended by the continuation
     * character (\).  Scans the script until it finds an un-escaped
     * newline.
     */
    private String readLine(String line, BufferedReader in) throws IOException {
      line = line.trim();
      boolean more = line.endsWith("\\");
      if (more) {
        line = line.substring(0, line.lastIndexOf('\\')); // snip
        StringBuilder buf = new StringBuilder(line);        // save
        while (more) {
          line = in.readLine();
          if (line == null) {
            break;
          }
          line = line.trim();
          more = line.endsWith("\\");
          if (more) {
            line = line.substring(0, line.lastIndexOf('\\'));
          }
          buf.append(' ').append(line);
        }
        line = buf.toString().trim();
      }

      if (scriptHasVars && line.contains("$")) {
        line = vars.expand(line);
      }

      return line;
    }

    /**
     * Returns a block of SQL, starting with the given String. Returns
     * <code> startOfSql</code> concatenated with each line from
     * <code>in</code> until a line ending with a semicolon is found.
     */
    private String readSql(String startOfSql, BufferedReader in)
        throws IOException {
      // REVIEW mb StringBuffer not always needed
      StringBuilder sql = new StringBuilder(startOfSql);
      sql.append('\n');

      String line;
      if (!startOfSql.trim().endsWith(";")) {
        while ((line = in.readLine()) != null) {
          sql.append(line).append('\n');
          if (line.trim().endsWith(";")) {
            break;
          }
        }
      }

      line = sql.toString().trim();
      if (scriptHasVars && line.contains("$")) {
        line = vars.expand(line);
      }
      return line;
    }
  }


  /** Print command.
   *
   * <p>When executed, a @print command defines how any following @fetch
   * or @select commands will handle their resuult rows. MTSQL can print all
   * rows, no rows, or every nth row. A printed row can be prefixed by a
   * sequence nuber and/or the time it was received (a different notion than
   * its rowtime, which often tells when it was inserted).
   */
  private class PrintCommand extends AbstractCommand {
    // print every nth row: 1 means all rows, 0 means no rows.
    private final int nth;
    private final boolean count;    // print a sequence number
    private final boolean time;     // print the time row was fetched
    // print total row count and elapsed fetch time:
    private final boolean total;
    // TODO: more control of formats

    PrintCommand(String spec) {
      int nth = 0;
      boolean count = false;
      boolean time = false;
      boolean total = false;
      StringTokenizer tokenizer = new StringTokenizer(spec);

      if (tokenizer.countTokens() == 0) {
        // a bare "@print" means "@print all"
        nth = 1;
      } else {
        while (tokenizer.hasMoreTokens()) {
          String token = tokenizer.nextToken();
          if (token.equalsIgnoreCase("none")) {
            nth = 0;
          } else if (token.equalsIgnoreCase("all")) {
            nth = 1;
          } else if (token.equalsIgnoreCase("total")) {
            total = true;
          } else if (token.equalsIgnoreCase("count")) {
            count = true;
          } else if (token.equalsIgnoreCase("time")) {
            time = true;
          } else if (token.equalsIgnoreCase("every")) {
            nth = 1;
            if (tokenizer.hasMoreTokens()) {
              token = tokenizer.nextToken();
              nth = Integer.parseInt(token);
            }
          }
        }
      }
      this.nth = nth;
      this.count = count;
      this.time = time;
      this.total = total;
    }

    protected void doExecute(ConcurrentTestCommandExecutor executor)
        throws SQLException {
      Integer threadId = executor.getThreadId();
      BufferedWriter out = threadBufferedWriters.get(threadId);
      threadResultsReaders.put(
          threadId, new ResultsReader(out, nth, count, time, total));
    }
  }

  /** Echo command. */
  private class EchoCommand extends AbstractCommand {
    private final String msg;

    private EchoCommand(String msg) {
      this.msg = msg;
    }

    protected void doExecute(ConcurrentTestCommandExecutor executor)
        throws SQLException {
      storeMessage(executor.getThreadId(), msg);
    }
  }

  /** Plugin command. */
  private class PluginCommand extends AbstractCommand {

    private final ConcurrentTestPluginCommand pluginCommand;

    private PluginCommand(
        String command,
        String params) throws IOException {
      ConcurrentTestPlugin plugin = pluginForCommand.get(command);
      pluginCommand = plugin.getCommandFor(command, params);
    }

    protected void doExecute(final ConcurrentTestCommandExecutor exec)
        throws Exception {
      ConcurrentTestPluginCommand.TestContext context =
          new ConcurrentTestPluginCommand.TestContext() {
            public void storeMessage(String message) {
              ConcurrentTestCommandScript.this.storeMessage(
                  exec.getThreadId(), message);
            }

            public Connection getConnection() {
              return exec.getConnection();
            }

            public Statement getCurrentStatement() {
              return exec.getStatement();
            }
          };
      pluginCommand.execute(context);
    }
  }

  // Matches shell wilcards and other special characters: when a command
  // contains some of these, it needs a shell to run it.
  private final Pattern shellWildcardPattern = Pattern.compile("[*?$|<>&]");

  // REVIEW mb 2/24/09 (Mardi Gras) Should this have a timeout?
  /** Shell command. */
  private class ShellCommand extends AbstractCommand {
    private final String command;
    private List<String> argv;      // the command, parsed and massaged

    private ShellCommand(String command) {
      this.command = command;
      boolean needShell = hasWildcard(command);
      if (needShell) {
        argv = new ArrayList<>();
        argv.add("/bin/sh");
        argv.add("-c");
        argv.add(command);
      } else {
        argv = tokenize(command);
      }
    }

    private boolean hasWildcard(String command) {
      return shellWildcardPattern.matcher(command).find();
    }

    private List<String> tokenize(String s) {
      List<String> result = new ArrayList<>();
      StringTokenizer tokenizer = new StringTokenizer(s);
      while (tokenizer.hasMoreTokens()) {
        result.add(tokenizer.nextToken());
      }
      return result;
    }

    protected void doExecute(ConcurrentTestCommandExecutor executor) {
      Integer threadId = executor.getThreadId();
      storeMessage(threadId, command);

      // argv[0] is found on $PATH. Working directory is the script's home
      // directory.
      //
      // WARNING: ProcessBuilder is security-sensitive. Its use is currently
      // safe because this code is under "core/test". Developers must not move
      // this code into "core/main".
      ProcessBuilder pb = new ProcessBuilder(argv);
      pb.directory(scriptDirectory);
      try {
        // direct stdout & stderr to the the threadWriter
        int status = runAppProcess(pb, null, null, getThreadWriter(threadId));
        if (status != 0) {
          storeMessage(threadId,
              "command " + command + ": exited with status " + status);
        }
      } catch (Exception e) {
        storeMessage(threadId,
            "command " + command + ": failed with exception " + e.getMessage());
      }
    }
  }

  /** Command that has a timeout. */
  // TODO: replace by super.CommmandWithTimeout
  private abstract static class CommandWithTimeout extends AbstractCommand {
    private long timeout;

    private CommandWithTimeout(long timeout) {
      this.timeout = timeout;
    }

    // returns the timeout as set (-1 means no timeout)
    protected long setTimeout(Statement stmt) throws SQLException {
      assert timeout >= 0;
      if (timeout > 0) {
        // FIX: call setQueryTimeoutMillis() when available.
        assert timeout >= 1000 : "timeout too short";
        int t = (int) (timeout / 1000);
        stmt.setQueryTimeout(t);
        return t;
      }
      return -1;
    }
  }

  /** Command with timeout and row limit. */
  private abstract static class CommandWithTimeoutAndRowLimit
      extends CommandWithTimeout {
    private int rowLimit;

    private CommandWithTimeoutAndRowLimit(long timeout) {
      this(timeout, 0);
    }

    private CommandWithTimeoutAndRowLimit(long timeout, int rowLimit) {
      super(timeout);
      this.rowLimit = rowLimit;
    }

    protected void setRowLimit(Statement stmt) throws SQLException {
      assert rowLimit >= 0;
      if (rowLimit > 0) {
        stmt.setMaxRows(rowLimit);
      }
    }
  }

  /**
   * SelectCommand creates and executes a SQL select statement, with optional
   * timeout and row limit.
   */
  private class SelectCommand extends CommandWithTimeoutAndRowLimit {
    private String sql;

    private SelectCommand(String sql) {
      this(sql, 0, 0);
    }

    private SelectCommand(String sql, boolean errorExpected) {
      this(sql, 0, 0);
      if (errorExpected) {
        this.markToFail();
      }
    }

    private SelectCommand(String sql, long timeout) {
      this(sql, timeout, 0);
    }

    private SelectCommand(String sql, long timeout, int rowLimit) {
      super(timeout, rowLimit);
      this.sql = sql;
    }

    protected void doExecute(ConcurrentTestCommandExecutor executor)
        throws SQLException {
      // TODO: trim and chop in constructor; stash sql in base class;
      // execute() calls storeSql.
      String properSql = sql.trim();

      storeSql(
          executor.getThreadId(),
          properSql);

      if (properSql.endsWith(";")) {
        properSql = properSql.substring(0, properSql.length() - 1);
      }

      PreparedStatement stmt =
          executor.getConnection().prepareStatement(properSql);
      long timeout = setTimeout(stmt);
      setRowLimit(stmt);

      try {
        storeResults(
            executor.getThreadId(),
            stmt.executeQuery(),
            timeout);
      } finally {
        stmt.close();
      }
    }
  }

  /**
   * SelectCommand creates and executes a SQL select statement, with optional
   * timeout.
   */
  private class SqlCommand extends CommandWithTimeout {
    private String sql;

    private SqlCommand(String sql) {
      super(0);
      this.sql = sql;
    }

    private SqlCommand(String sql, boolean errorExpected) {
      super(0);
      this.sql = sql;
      if (errorExpected) {
        this.markToFail();
      }
    }

    private SqlCommand(String sql, long timeout) {
      super(timeout);
      this.sql = sql;
    }

    protected void doExecute(ConcurrentTestCommandExecutor executor)
        throws SQLException {
      String properSql = sql.trim();

      storeSql(
          executor.getThreadId(),
          properSql);

      if (properSql.endsWith(";")) {
        properSql = properSql.substring(0, properSql.length() - 1);
      }

      if (properSql.equalsIgnoreCase("commit")) {
        executor.getConnection().commit();
        return;
      } else if (properSql.equalsIgnoreCase("rollback")) {
        executor.getConnection().rollback();
        return;
      }

      PreparedStatement stmt =
          executor.getConnection().prepareStatement(properSql);
      long timeout = setTimeout(stmt);
      boolean timeoutSet = timeout >= 0;

      try {
        boolean haveResults = stmt.execute();
        if (haveResults) {
          // Farrago rewrites "call" statements as selects.
          storeMessage(
              executor.getThreadId(),
              "0 rows affected.");
          // is there anything interesting in the ResultSet?
        } else {
          int rows = stmt.getUpdateCount();
          if (rows != 1) {
            storeMessage(
                executor.getThreadId(),
                String.valueOf(rows) + " rows affected.");
          } else {
            storeMessage(
                executor.getThreadId(),
                "1 row affected.");
          }
        }
      } catch (SqlTimeoutException e) {
        if (!timeoutSet) {
          throw e;
        }

        Util.swallow(e, null);
        storeMessage(
            executor.getThreadId(),
            "Timeout");
      } finally {
        stmt.close();
      }
    }
  }

  /**
   * PrepareCommand creates a {@link PreparedStatement}, which is saved as the
   * current statement of its test thread. For a preparted query (a SELECT or
   * a CALL with results), a subsequent FetchAndPrintCommand executes the
   * statement and fetches its reults, until end-of-data or a timeout. A
   * PrintCommand attaches a listener, called for each rows, that selects rows
   * to save and print, and sets the format. By default, if no PrintCommand
   * appears before a FetchAndPrintCommand, all rows are printed. A
   * CloseCommand closes and discards the prepared statement.
   */
  private class PrepareCommand extends AbstractCommand {
    private String sql;

    private PrepareCommand(String sql) {
      this.sql = sql;
    }

    protected void doExecute(ConcurrentTestCommandExecutor executor)
        throws SQLException {
      String properSql = sql.trim();

      storeSql(
          executor.getThreadId(),
          properSql);

      if (properSql.endsWith(";")) {
        properSql = properSql.substring(0, properSql.length() - 1);
      }

      PreparedStatement stmt =
          executor.getConnection().prepareStatement(properSql);

      executor.setStatement(stmt);
    }
  }

  /**
   * FetchAndPrintCommand executes a previously prepared statement stored
   * inthe ConcurrentTestCommandExecutor and then outputs the returned
   * rows.
   */
  private class FetchAndPrintCommand extends CommandWithTimeout {
    private FetchAndPrintCommand(long timeout) {
      super(timeout);
    }

    protected void doExecute(ConcurrentTestCommandExecutor executor)
        throws SQLException {
      PreparedStatement stmt =
          (PreparedStatement) executor.getStatement();
      long timeout = setTimeout(stmt);

      storeResults(
          executor.getThreadId(),
          stmt.executeQuery(),
          timeout);
    }
  }

  /** Result reader. */
  private class ResultsReader {
    private final PrintWriter out;
    // print every Nth row. 1 means all rows, 0 means none.
    private final int nth;
    // prefix printed row with its sequence number
    private final boolean counted;
    // prefix printed row with time it was fetched
    private final boolean timestamped;
    // print final summary, rows & elapsed time.
    private final boolean totaled;

    private long baseTime = 0;
    private int rowCount = 0;
    private int ncols = 0;
    private int[] widths;
    private String[] labels;

    ResultsReader(BufferedWriter out) {
      this(out, 1, false, false, false);
    }

    ResultsReader(
        BufferedWriter out,
        int nth, boolean counted, boolean timestamped, boolean totaled) {
      this.out = new PrintWriter(out);
      this.nth = nth;
      this.counted = counted;
      this.timestamped = timestamped;
      this.totaled = totaled;
      this.baseTime = scriptStartTime;
    }

    void prepareFormat(ResultSet rset) throws SQLException {
      ResultSetMetaData meta = rset.getMetaData();
      ncols = meta.getColumnCount();
      widths = new int[ncols];
      labels = new String[ncols];
      for (int i = 0; i < ncols; i++) {
        labels[i] = meta.getColumnLabel(i + 1);
        int displaySize = meta.getColumnDisplaySize(i + 1);

        // NOTE jvs 13-June-2006: I put this in to cap EXPLAIN PLAN,
        // which now returns a very large worst-case display size.
        if (displaySize > 4096) {
          displaySize = 0;
        }
        widths[i] = Math.max(labels[i].length(), displaySize);
      }
    }

    private void printHeaders() {
      printSeparator();
      indent();
      printRow(labels);
      printSeparator();
    }

    void read(ResultSet rset, long timeout) throws SQLException {
      boolean withTimeout = timeout >= 0;
      boolean timedOut = false;
      long startTime = 0;
      long endTime = 0;
      try {
        prepareFormat(rset);
        String[] values = new String[ncols];
        int printedRowCount = 0;
        if (nth > 0) {
          printHeaders();
        }
        startTime = System.currentTimeMillis();
        for (rowCount = 0; rset.next(); rowCount++) {
          if (nth == 0) {
            continue;
          }
          if (nth == 1 || rowCount % nth == 0) {
            long time = System.currentTimeMillis();
            if (printedRowCount > 0
                && (printedRowCount % 100 == 0)) {
              printHeaders();
            }
            for (int i = 0; i < ncols; i++) {
              values[i] = rset.getString(i + 1);
            }
            if (counted) {
              printRowCount(rowCount);
            }
            if (timestamped) {
              printTimestamp(time);
            }
            printRow(values);
            printedRowCount++;
          }
        }
      } catch (SqlTimeoutException e) {
        endTime = System.currentTimeMillis();
        timedOut = true;
        if (!withTimeout) {
          throw e;
        }
        Util.swallow(e, null);
      } catch (SQLException e) {
        endTime = System.currentTimeMillis();
        timedOut = true;

        // 2007-10-23 hersker: hack to ignore timeout exceptions
        // from other Farrago projects without being able to
        // import/reference the actual exceptions
        final String eClassName = e.getClass().getName();
        if (eClassName.endsWith("TimeoutException")) {
          if (!withTimeout) {
            throw e;
          }
          Util.swallow(e, null);
        } else {
          Util.swallow(e, null);
          out.println(e.getMessage());
        }
      } catch (RuntimeException e) {
        e.printStackTrace();
        throw e;
      } finally {
        if (endTime == 0) {
          endTime = System.currentTimeMillis();
        }
        rset.close();
        if (nth > 0) {
          printSeparator();
          out.println();
        }
        if (verbose) {
          out.printf(Locale.ROOT, "fetch started at %tc %d, %s at %tc %d%n",
              startTime, startTime,
              timedOut ? "timeout" : "eos",
              endTime, endTime);
        }
        if (totaled) {
          long dt = endTime - startTime;
          if (withTimeout) {
            dt -= timeout;
          }
          assert dt >= 0;
          out.printf(Locale.ROOT, "fetched %d rows in %d msecs %s%n",
              rowCount, dt, timedOut ? "(timeout)" : "(end)");
        }
      }
    }

    private void printRowCount(int count) {
      out.printf(Locale.ROOT, "(%06d) ", count);
    }

    private void printTimestamp(long time) {
      time -= baseTime;
      out.printf(Locale.ROOT, "(% 4d.%03d) ", time / 1000, time % 1000);
    }

    // indent a heading or separator line to match a row-values line
    private void indent() {
      if (counted) {
        out.print("         ");
      }
      if (timestamped) {
        out.print("           ");
      }
    }

    /**
     * Prints an output table separator. Something like <code>
     * "+----+--------+"</code>.
     */
    private void printSeparator() {
      indent();
      for (int i = 0; i < widths.length; i++) {
        if (i > 0) {
          out.write("-+-");
        } else {
          out.write("+-");
        }

        int numDashes = widths[i];
        while (numDashes > 0) {
          out.write(
              DASHES,
              0,
              Math.min(numDashes, BUF_SIZE));
          numDashes -= Math.min(numDashes, BUF_SIZE);
        }
      }
      out.println("-+");
    }

    /**
     * Prints an output table row. Something like <code>"| COL1 | COL2
     * |"</code>.
     */
    private void printRow(String[] values) {
      for (int i = 0; i < values.length; i++) {
        String value = values[i];
        if (value == null) {
          value = "";
        }
        if (i > 0) {
          out.write(" | ");
        } else {
          out.write("| ");
        }
        out.write(value);
        int excess = widths[i] - value.length();
        while (excess > 0) {
          out.write(
              SPACES,
              0,
              Math.min(excess, BUF_SIZE));
          excess -= Math.min(excess, BUF_SIZE);
        }
      }
      out.println(" |");
    }
  }


  /** Standalone client test tool. */
  private static class Tool {
    boolean quiet = false;          // -q
    boolean verbose = false;        // -v
    boolean debug = false;          // -g
    String server;                  // -u
    String driver;                  // -d
    String user;                    // -n
    String password;                // -p
    List<String> bindings;          // VAR=VAL
    List<String> files;             // FILE

    Tool() {
      bindings = new ArrayList<>();
      files = new ArrayList<>();
    }

    // returns 0 on success, 1 on error, 2 on bad invocation.
    public int run(String[] args) {
      try (final PrintWriter w = Util.printWriter(System.out)) {
        if (!parseCommand(args)) {
          usage();
          return 2;
        }

        Class z = Class.forName(driver); // load driver
        Properties jdbcProps = new Properties();
        if (user != null) {
          jdbcProps.setProperty("user", user);
        }
        if (password != null) {
          jdbcProps.setProperty("password", password);
        }

        for (String file : files) {
          ConcurrentTestCommandScript script =
              new ConcurrentTestCommandScript();
          try {
            script.setQuiet(quiet);
            script.setVerbose(verbose);
            script.setDebug(debug);
            script.prepare(file, bindings);
            script.setDataSource(server, jdbcProps);
            script.execute();
          } finally {
            if (!quiet) {
              script.printResults(w);
            }
          }
        }
      } catch (Exception e) {
        System.err.println(e.getMessage());
        return 1;
      }
      return 0;
    }

    static void usage() {
      System.err.println(
          "Usage: mtsql [-vg] -u SERVER -d DRIVER "
          + "[-n USER][-p PASSWORD] SCRIPT [SCRIPT]...");
    }

    boolean parseCommand(String[] args) {
      try {
        // very permissive as to order
        for (int i = 0; i < args.length;) {
          String arg = args[i++];
          if (arg.charAt(0) == '-') {
            switch (arg.charAt(1)) {
            case 'v':
              verbose = true;
              break;
            case 'q':
              quiet = true;
              break;
            case 'g':
              debug = true;
              break;
            case 'u':
              this.server = args[i++];
              break;
            case 'd':
              this.driver = args[i++];
              break;
            case 'n':
              this.user = args[i++];
              break;
            case 'p':
              this.password = args[i++];
              break;
            default:
              return false;
            }
          } else if (arg.contains("=")) {
            if (Character.isJavaIdentifierStart(arg.charAt(0))) {
              bindings.add(arg);
            } else {
              return false;
            }
          } else {
            files.add(arg);
          }
        }
        if (server == null || driver == null) {
          return false;
        }
      } catch (Throwable th) {
        return false;
      }
      return true;
    }
  }

  /**
   * Client tool that connects via jdbc and runs one or more mtsql on that
   * connection.
   *
   * <p>Usage: mtsql [-vgq] -u SERVER -d DRIVER [-n USER][-p PASSWORD]
   * [VAR=VAL]...  SCRIPT [SCRIPT]...
   */
  public static void main(String[] args) {
    int status = new Tool().run(args);
    Unsafe.systemExit(status);
  }
}

// End ConcurrentTestCommandScript.java
