/**
 * 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.tajo;

import com.google.common.base.Charsets;
import com.google.common.io.Closeables;
import com.google.common.io.Files;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.HdfsConfiguration;
import org.apache.hadoop.hdfs.MiniDFSCluster;
import org.apache.hadoop.util.ShutdownHookManager;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.apache.tajo.catalog.*;
import org.apache.tajo.client.TajoClient;
import org.apache.tajo.client.TajoClientImpl;
import org.apache.tajo.client.TajoClientUtil;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.conf.TajoConf.ConfVars;
import org.apache.tajo.engine.planner.global.rewriter.GlobalPlanTestRuleProvider;
import org.apache.tajo.exception.UnsupportedCatalogStore;
import org.apache.tajo.master.TajoMaster;
import org.apache.tajo.plan.rewrite.LogicalPlanTestRuleProvider;
import org.apache.tajo.querymaster.Query;
import org.apache.tajo.querymaster.QueryMasterTask;
import org.apache.tajo.querymaster.Stage;
import org.apache.tajo.querymaster.StageState;
import org.apache.tajo.service.ServiceTrackerFactory;
import org.apache.tajo.storage.FileTablespace;
import org.apache.tajo.storage.TablespaceManager;
import org.apache.tajo.util.CommonTestingUtil;
import org.apache.tajo.util.KeyValueSet;
import org.apache.tajo.util.Pair;
import org.apache.tajo.util.history.QueryHistory;
import org.apache.tajo.worker.TajoWorker;

import java.io.File;
import java.io.IOException;
import java.io.Writer;
import java.net.InetSocketAddress;
import java.net.URI;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.TimeZone;
import java.util.UUID;

public class TajoTestingCluster {
	private static Log LOG = LogFactory.getLog(TajoTestingCluster.class);
	private TajoConf conf;
  private FileSystem defaultFS;
  private MiniDFSCluster dfsCluster;
	private CatalogServer catalogServer;
  private HBaseTestClusterUtil hbaseUtil;

  private TajoMaster tajoMaster;
  private List<TajoWorker> tajoWorkers = new ArrayList<TajoWorker>();
  private boolean isDFSRunning = false;
  private boolean isTajoClusterRunning = false;
  private boolean isCatalogServerRunning = false;

	private File clusterTestBuildDir = null;

	/**
	 * Default parent directory for test output.
	 */
	public static final String DEFAULT_TEST_DIRECTORY = "target/" + 
	    System.getProperty("tajo.test.data.dir", "test-data");

  /**
   * True If HiveCatalogStore is used. Otherwise, it is FALSE.
   */
  public Boolean isHiveCatalogStoreUse = false;

  private static final String LOG_LEVEL;

  static {
    LOG_LEVEL = System.getProperty("LOG_LEVEL");
  }

  public TajoTestingCluster() {
    this(false);
  }

  public TajoTestingCluster(boolean masterHaEMode) {
    this.conf = new TajoConf();
    this.conf.setBoolVar(ConfVars.TAJO_MASTER_HA_ENABLE, masterHaEMode);

    initTestDir();
    setTestingFlagProperties();
    initPropertiesAndConfigs();
  }

  void setTestingFlagProperties() {
    System.setProperty(TajoConstants.TEST_KEY, Boolean.TRUE.toString());
    conf.set(TajoConstants.TEST_KEY, Boolean.TRUE.toString());
  }

  void initPropertiesAndConfigs() {

    // Set time zone
    TimeZone testDefaultTZ = TimeZone.getTimeZone(TajoConstants.DEFAULT_SYSTEM_TIMEZONE);
    conf.setSystemTimezone(testDefaultTZ);
    TimeZone.setDefault(testDefaultTZ);

    // Injection of equality testing code of logical plan (de)serialization
    conf.setClassVar(ConfVars.LOGICAL_PLAN_REWRITE_RULE_PROVIDER_CLASS, LogicalPlanTestRuleProvider.class);
    conf.setClassVar(ConfVars.GLOBAL_PLAN_REWRITE_RULE_PROVIDER_CLASS, GlobalPlanTestRuleProvider.class);

    conf.setInt(ConfVars.WORKER_RESOURCE_AVAILABLE_CPU_CORES.varname, 4);
    conf.setInt(ConfVars.WORKER_RESOURCE_AVAILABLE_MEMORY_MB.varname, 2000);
    conf.setInt(ConfVars.WORKER_RESOURCE_AVAILABLE_DISK_PARALLEL_NUM.varname, 3);
    conf.setInt(ConfVars.SHUFFLE_FETCHER_PARALLEL_EXECUTION_MAX_NUM.varname, 2);

    // Client API RPC
    conf.setIntVar(ConfVars.RPC_CLIENT_WORKER_THREAD_NUM, 2);

    //Client API service RPC Server
    conf.setIntVar(ConfVars.MASTER_SERVICE_RPC_SERVER_WORKER_THREAD_NUM, 2);
    conf.setIntVar(ConfVars.WORKER_SERVICE_RPC_SERVER_WORKER_THREAD_NUM, 2);
    conf.setIntVar(ConfVars.REST_SERVICE_RPC_SERVER_WORKER_THREAD_NUM, 2);

    // Internal RPC Client
    conf.setIntVar(ConfVars.INTERNAL_RPC_CLIENT_WORKER_THREAD_NUM, 2);
    conf.setIntVar(ConfVars.SHUFFLE_RPC_CLIENT_WORKER_THREAD_NUM, 2);

    // Internal RPC Server
    conf.setIntVar(ConfVars.MASTER_RPC_SERVER_WORKER_THREAD_NUM, 2);
    conf.setIntVar(ConfVars.QUERY_MASTER_RPC_SERVER_WORKER_THREAD_NUM, 2);
    conf.setIntVar(ConfVars.WORKER_RPC_SERVER_WORKER_THREAD_NUM, 2);
    conf.setIntVar(ConfVars.CATALOG_RPC_SERVER_WORKER_THREAD_NUM, 2);
    conf.setIntVar(ConfVars.SHUFFLE_RPC_SERVER_WORKER_THREAD_NUM, 2);

    // Memory cache termination
    conf.setIntVar(ConfVars.HISTORY_QUERY_CACHE_SIZE, 10);

    // Python function path
    conf.setStrings(ConfVars.PYTHON_CODE_DIR.varname, getClass().getResource("/python").toString());

    // Query output file
    conf.setVar(ConfVars.QUERY_OUTPUT_DEFAULT_FILE_FORMAT, BuiltinStorages.DRAW);

    /* Since Travis CI limits the size of standard output log up to 4MB */
    if (!StringUtils.isEmpty(LOG_LEVEL)) {
      Level defaultLevel = Logger.getRootLogger().getLevel();
      Logger.getLogger("org.apache.tajo").setLevel(Level.toLevel(LOG_LEVEL.toUpperCase(), defaultLevel));
      Logger.getLogger("org.apache.hadoop").setLevel(Level.toLevel(LOG_LEVEL.toUpperCase(), defaultLevel));
      Logger.getLogger("org.apache.zookeeper").setLevel(Level.toLevel(LOG_LEVEL.toUpperCase(), defaultLevel));
      Logger.getLogger("BlockStateChange").setLevel(Level.toLevel(LOG_LEVEL.toUpperCase(), defaultLevel));
      Logger.getLogger("org.mortbay.log").setLevel(Level.toLevel(LOG_LEVEL.toUpperCase(), defaultLevel));
    }
  }

	public TajoConf getConfiguration() {
		return this.conf;
	}

  public void initTestDir() {
    if (clusterTestBuildDir == null) {
      clusterTestBuildDir = setupClusterTestBuildDir();
    }
  }

	/**
	 * @return Where to write test data on local filesystem; usually
	 * {@link #DEFAULT_TEST_DIRECTORY}
	 * @see #setupClusterTestBuildDir()
	 */
	public File getTestDir() {
		return clusterTestBuildDir;
	}

	/**
	 * @param subdirName
	 * @return Path to a subdirectory named <code>subdirName</code> under
	 * {@link #getTestDir()}.
	 * @see #setupClusterTestBuildDir()
	 */
	public static File getTestDir(final String subdirName) {
		return new File(new File(DEFAULT_TEST_DIRECTORY), subdirName);
  }

	public static File setupClusterTestBuildDir() {
		String randomStr = UUID.randomUUID().toString();
		String dirStr = getTestDir(randomStr).toString();
		File dir = new File(dirStr).getAbsoluteFile();
		// Have it cleaned up on exit
		dir.deleteOnExit();
		return dir;
	}

  ////////////////////////////////////////////////////////
  // HDFS Section
  ////////////////////////////////////////////////////////
  /**
   * Start a minidfscluster.
   * @param servers How many DNs to start.
   * @throws Exception
   * @see {@link #shutdownMiniDFSCluster()}
   * @return The mini dfs cluster created.
   */
  public MiniDFSCluster startMiniDFSCluster(int servers) throws Exception {
    return startMiniDFSCluster(servers, null, null);
  }

  /**
   * Start a minidfscluster.
   * Can only create one.
   * @param servers How many DNs to start.
   * @param dir Where to home your dfs cluster.
   * @param hosts hostnames DNs to run on.
   * @throws Exception
   * @see {@link #shutdownMiniDFSCluster()}
   * @return The mini dfs cluster created.
   * @throws java.io.IOException
   */
  public MiniDFSCluster startMiniDFSCluster(int servers,
                                            File dir,
                                            final String hosts[])
      throws IOException {

    conf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, dir.toString());
    conf.setInt(DFSConfigKeys.DFS_REPLICATION_KEY, 1);
    conf.setBoolean(DFSConfigKeys.DFS_CLIENT_READ_SHORTCIRCUIT_KEY, false);
    MiniDFSCluster.Builder builder = new MiniDFSCluster.Builder(new HdfsConfiguration(conf));
    builder.hosts(hosts);
    builder.numDataNodes(servers);
    builder.format(true);
    builder.manageNameDfsDirs(true);
    builder.manageDataDfsDirs(true);
    builder.waitSafeMode(true);
    this.dfsCluster = builder.build();

    // Set this just-started cluster as our filesystem.
    this.defaultFS = this.dfsCluster.getFileSystem();
    this.conf.set(CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY, defaultFS.getUri().toString());
    this.conf.setVar(TajoConf.ConfVars.ROOT_DIR, defaultFS.getUri() + "/tajo");
    isDFSRunning = true;
    return this.dfsCluster;
  }

  public void shutdownMiniDFSCluster() throws Exception {
    if (this.dfsCluster != null) {
      try {
        FileSystem fs = this.dfsCluster.getFileSystem();
        if (fs != null) fs.close();
      } catch (IOException e) {
        System.err.println("error closing file system: " + e);
      }
      // The below throws an exception per dn, AsynchronousCloseException.
      this.dfsCluster.shutdown();
    }
  }

  public boolean isRunningDFSCluster() {
    return this.defaultFS != null;
  }

  public MiniDFSCluster getMiniDFSCluster() {
    return this.dfsCluster;
  }

  public FileSystem getDefaultFileSystem() {
    return this.defaultFS;
  }

  public HBaseTestClusterUtil getHBaseUtil() {
    return hbaseUtil;
  }

  ////////////////////////////////////////////////////////
  // Catalog Section
  ////////////////////////////////////////////////////////
  public CatalogServer startCatalogCluster() throws Exception {
    if(isCatalogServerRunning) throw new IOException("Catalog Cluster already running");

    CatalogTestingUtil.configureCatalog(conf, clusterTestBuildDir.getAbsolutePath());
    LOG.info("Apache Derby repository is set to " + conf.get(CatalogConstants.CATALOG_URI));
    conf.setVar(ConfVars.CATALOG_ADDRESS, "localhost:0");

    catalogServer = new CatalogServer();
    catalogServer.init(conf);
    catalogServer.start();
    isCatalogServerRunning = true;
    return this.catalogServer;
  }

  public void shutdownCatalogCluster() {
    if (catalogServer != null) {
      this.catalogServer.stop();
    }
    isCatalogServerRunning = false;
  }

  public CatalogServer getMiniCatalogCluster() {
    return this.catalogServer;
  }

  public CatalogService getCatalogService() {
    return new LocalCatalogWrapper(catalogServer);
  }

  public boolean isHiveCatalogStoreRunning() {
    return isHiveCatalogStoreUse;
  }

  ////////////////////////////////////////////////////////
  // Tajo Cluster Section
  ////////////////////////////////////////////////////////
  private void startMiniTajoCluster(File testBuildDir,
                                    final int numSlaves,
                                    boolean local) throws Exception {
    TajoConf c = getConfiguration();
    c.setVar(ConfVars.TAJO_MASTER_CLIENT_RPC_ADDRESS, "localhost:0");
    c.setVar(ConfVars.TAJO_MASTER_UMBILICAL_RPC_ADDRESS, "localhost:0");
    c.setVar(ConfVars.RESOURCE_TRACKER_RPC_ADDRESS, "localhost:0");
    c.setVar(ConfVars.WORKER_PEER_RPC_ADDRESS, "localhost:0");
    c.setVar(ConfVars.WORKER_TEMPORAL_DIR, "file://" + testBuildDir.getAbsolutePath() + "/tajo-localdir");
    c.setIntVar(ConfVars.REST_SERVICE_PORT, 0);

    if (!local) {
      String tajoRootDir = getMiniDFSCluster().getFileSystem().getUri().toString() + "/tajo";
      c.setVar(ConfVars.ROOT_DIR, tajoRootDir);

      URI defaultTsUri = TajoConf.getWarehouseDir(c).toUri();
      FileTablespace defaultTableSpace =
          new FileTablespace(TablespaceManager.DEFAULT_TABLESPACE_NAME, defaultTsUri, null);
      defaultTableSpace.init(conf);
      TablespaceManager.addTableSpaceForTest(defaultTableSpace);

    } else {
      c.setVar(ConfVars.ROOT_DIR, "file://" + testBuildDir.getAbsolutePath() + "/tajo");
    }

    setupCatalogForTesting(c, testBuildDir);

    LOG.info("derby repository is set to " + conf.get(CatalogConstants.CATALOG_URI));

    tajoMaster = new TajoMaster();
    tajoMaster.init(c);
    tajoMaster.start();

    this.conf.setVar(ConfVars.WORKER_PEER_RPC_ADDRESS, c.getVar(ConfVars.WORKER_PEER_RPC_ADDRESS));
    this.conf.setVar(ConfVars.TAJO_MASTER_CLIENT_RPC_ADDRESS, c.getVar(ConfVars.TAJO_MASTER_CLIENT_RPC_ADDRESS));

    InetSocketAddress tajoMasterAddress = tajoMaster.getContext().getTajoMasterService().getBindAddress();

    this.conf.setVar(ConfVars.TAJO_MASTER_UMBILICAL_RPC_ADDRESS,
        tajoMasterAddress.getHostName() + ":" + tajoMasterAddress.getPort());
    this.conf.setVar(ConfVars.RESOURCE_TRACKER_RPC_ADDRESS, c.getVar(ConfVars.RESOURCE_TRACKER_RPC_ADDRESS));
    this.conf.setVar(ConfVars.CATALOG_ADDRESS, c.getVar(ConfVars.CATALOG_ADDRESS));
    
    InetSocketAddress tajoRestAddress = tajoMaster.getContext().getRestServer().getBindAddress();
    
    this.conf.setIntVar(ConfVars.REST_SERVICE_PORT, tajoRestAddress.getPort());

    startTajoWorkers(numSlaves);

    isTajoClusterRunning = true;
    LOG.info("Mini Tajo cluster is up");
    LOG.info("====================================================================================");
    LOG.info("=                           MiniTajoCluster starts up                              =");
    LOG.info("====================================================================================");
    LOG.info("= * Master Address: " + tajoMaster.getMasterName());
    LOG.info("= * CatalogStore: " + tajoMaster.getCatalogServer().getStoreClassName());
    LOG.info("------------------------------------------------------------------------------------");
    LOG.info("= * Warehouse Dir: " + TajoConf.getWarehouseDir(c));
    LOG.info("= * Worker Tmp Dir: " + c.getVar(ConfVars.WORKER_TEMPORAL_DIR));
    LOG.info("====================================================================================");
  }

  private void setupCatalogForTesting(TajoConf c, File testBuildDir) throws IOException, UnsupportedCatalogStore {
    final String HIVE_CATALOG_CLASS_NAME = "org.apache.tajo.catalog.store.HiveCatalogStore";
    boolean hiveCatalogClassExists = false;
    try {
      getClass().getClassLoader().loadClass(HIVE_CATALOG_CLASS_NAME);
      hiveCatalogClassExists = true;
    } catch (ClassNotFoundException e) {
      LOG.info("HiveCatalogStore is not available.");
    }
    String driverClass = System.getProperty(CatalogConstants.STORE_CLASS);

    if (hiveCatalogClassExists &&
        driverClass != null && driverClass.equals(HIVE_CATALOG_CLASS_NAME)) {
      try {
        getClass().getClassLoader().loadClass(HIVE_CATALOG_CLASS_NAME);
        String jdbcUri = "jdbc:derby:;databaseName="+ testBuildDir.toURI().getPath()  + "/metastore_db;create=true";
        c.set("hive.metastore.warehouse.dir", TajoConf.getWarehouseDir(c).toString() + "/default");
        c.set("javax.jdo.option.ConnectionURL", jdbcUri);
        c.set(TajoConf.ConfVars.WAREHOUSE_DIR.varname, conf.getVar(ConfVars.WAREHOUSE_DIR));
        c.set(CatalogConstants.STORE_CLASS, HIVE_CATALOG_CLASS_NAME);
        Path defaultDatabasePath = new Path(TajoConf.getWarehouseDir(c).toString() + "/default");
        FileSystem fs = defaultDatabasePath.getFileSystem(c);
        if (!fs.exists(defaultDatabasePath)) {
          fs.mkdirs(defaultDatabasePath);
        }
        isHiveCatalogStoreUse = true;
      } catch (ClassNotFoundException cnfe) {
        throw new IOException(cnfe);
      }
    } else { // for derby
      CatalogTestingUtil.configureCatalog(conf, testBuildDir.getAbsolutePath());
    }
    c.setVar(ConfVars.CATALOG_ADDRESS, "localhost:0");
  }

  private void startTajoWorkers(int numSlaves) throws Exception {
    for(int i = 0; i < 1; i++) {
      TajoWorker tajoWorker = new TajoWorker();

      TajoConf workerConf  = new TajoConf(this.conf);

      workerConf.setVar(ConfVars.WORKER_INFO_ADDRESS, "localhost:0");
      workerConf.setVar(ConfVars.WORKER_CLIENT_RPC_ADDRESS, "localhost:0");
      workerConf.setVar(ConfVars.WORKER_PEER_RPC_ADDRESS, "localhost:0");

      workerConf.setVar(ConfVars.WORKER_QM_RPC_ADDRESS, "localhost:0");
      
      tajoWorker.startWorker(workerConf, new String[0]);

      LOG.info("MiniTajoCluster Worker #" + (i + 1) + " started.");
      tajoWorkers.add(tajoWorker);
    }
  }

  public TajoMaster getMaster() {
    return this.tajoMaster;
  }

  public List<TajoWorker> getTajoWorkers() {
    return this.tajoWorkers;
  }

  public void shutdownMiniTajoCluster() {
    if(this.tajoMaster != null) {
      this.tajoMaster.stop();
    }
    for(TajoWorker eachWorker: tajoWorkers) {
      eachWorker.stopWorkerForce();
    }
    tajoWorkers.clear();
    this.tajoMaster= null;
  }

  ////////////////////////////////////////////////////////
  // Meta Cluster Section
  ////////////////////////////////////////////////////////
  /**
   * @throws java.io.IOException If a cluster -- dfs or engine -- already running.
   */
  void isRunningCluster() throws IOException {
    if (!isTajoClusterRunning && !isCatalogServerRunning && !isDFSRunning) return;
    throw new IOException("Cluster already running at " +
        this.clusterTestBuildDir);
  }

  /**
   * This method starts up a tajo cluster with a given number of clusters in
   * distributed mode.
   *
   * @param numSlaves the number of tajo cluster to start up
   * @throws Exception
   */
  public void startMiniCluster(final int numSlaves)
      throws Exception {
    startMiniCluster(numSlaves, null);
  }

  public void startMiniCluster(final int numSlaves, final String [] dataNodeHosts) throws Exception {

    int numDataNodes = numSlaves;
    if(dataNodeHosts != null && dataNodeHosts.length != 0) {
      numDataNodes = dataNodeHosts.length;
    }

    LOG.info("Starting up minicluster with 1 master(s) and " +
        numSlaves + " worker(s) and " + numDataNodes + " datanode(s)");

    // If we already bring up the cluster, fail.
    isRunningCluster();
    if (clusterTestBuildDir != null) {
      LOG.info("Using passed path: " + clusterTestBuildDir);
    }

    startMiniDFSCluster(numDataNodes, clusterTestBuildDir, dataNodeHosts);
    this.dfsCluster.waitClusterUp();

    conf.setInt("hbase.hconnection.threads.core", 5);
    conf.setInt("hbase.hconnection.threads.max", 50);
    hbaseUtil = new HBaseTestClusterUtil(conf, clusterTestBuildDir);

    startMiniTajoCluster(this.clusterTestBuildDir, numSlaves, false);
  }

  public void startMiniClusterInLocal(final int numSlaves) throws Exception {
    isRunningCluster();

    if (clusterTestBuildDir != null) {
      LOG.info("Using passed path: " + clusterTestBuildDir);
    }

    startMiniTajoCluster(this.clusterTestBuildDir, numSlaves, true);
  }

  public void shutdownMiniCluster() throws IOException {
    LOG.info("========================================");
    LOG.info("Minicluster is stopping");
    LOG.info("========================================");

    try {
      Thread.sleep(3000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }

    shutdownMiniTajoCluster();

    if(this.catalogServer != null) {
      shutdownCatalogCluster();
      isCatalogServerRunning = false;
    }

    try {
      Thread.sleep(3000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }

    if(this.dfsCluster != null) {
      try {
        FileSystem fs = this.dfsCluster.getFileSystem();
        if (fs != null) fs.close();
        this.dfsCluster.shutdown();
      } catch (IOException e) {
        System.err.println("error closing file system: " + e);
      }
      isDFSRunning = false;
    }

    if(this.clusterTestBuildDir != null && this.clusterTestBuildDir.exists()) {
      if(!ShutdownHookManager.get().isShutdownInProgress()) {
        //TODO clean test dir when ShutdownInProgress
        LocalFileSystem localFS = LocalFileSystem.getLocal(conf);
        localFS.delete(new Path(clusterTestBuildDir.toString()), true);
        localFS.close();
      }
      this.clusterTestBuildDir = null;
    }

    if(hbaseUtil != null) {
      hbaseUtil.stopZooKeeperCluster();
      hbaseUtil.stopHBaseCluster();
    }

    LOG.info("Minicluster is down");
    isTajoClusterRunning = false;
  }

  public TajoClient newTajoClient() throws Exception {
    return new TajoClientImpl(ServiceTrackerFactory.get(getConfiguration()));
  }

  public static ResultSet run(String[] names,
                              Schema[] schemas,
                              KeyValueSet tableOption,
                              String[][] tables,
                              String query,
                              TajoClient client) throws Exception {
    TajoTestingCluster util = TpchTestBase.getInstance().getTestingCluster();

    FileSystem fs = util.getDefaultFileSystem();
    Path rootDir = TajoConf.getWarehouseDir(util.getConfiguration());
    fs.mkdirs(rootDir);
    for (int i = 0; i < names.length; i++) {
      createTable(names[i], schemas[i], tableOption, tables[i]);
    }
    Thread.sleep(1000);
    ResultSet res = client.executeQueryAndGetResult(query);
    return res;
  }

  public static ResultSet run(String[] names,
                              Schema[] schemas,
                              KeyValueSet tableOption,
                              String[][] tables,
                              String query) throws Exception {
    TpchTestBase instance = TpchTestBase.getInstance();
    TajoTestingCluster util = instance.getTestingCluster();
    while(true) {
      if(util.getMaster().isMasterRunning()) {
        break;
      }
      Thread.sleep(1000);
    }
    TajoConf conf = util.getConfiguration();
    TajoClient client = new TajoClientImpl(ServiceTrackerFactory.get(conf));

    try {
      return run(names, schemas, tableOption, tables, query, client);
    } finally {
      client.close();
    }
  }

  public static TajoClient newTajoClient(TajoTestingCluster util) throws SQLException, InterruptedException {
    while(true) {
      if(util.getMaster().isMasterRunning()) {
        break;
      }
      Thread.sleep(1000);
    }
    TajoConf conf = util.getConfiguration();
    return new TajoClientImpl(ServiceTrackerFactory.get(conf));
  }

  public static void createTable(String tableName, Schema schema,
                                 KeyValueSet tableOption, String[] tableDatas) throws Exception {
    createTable(tableName, schema, tableOption, tableDatas, 1);
  }

  public static void createTable(String tableName, Schema schema,
                                 KeyValueSet tableOption, String[] tableDatas, int numDataFiles) throws Exception {
    TpchTestBase instance = TpchTestBase.getInstance();
    TajoTestingCluster util = instance.getTestingCluster();
    TajoClient client = newTajoClient(util);
    try {
      FileSystem fs = util.getDefaultFileSystem();
      Path rootDir = TajoConf.getWarehouseDir(util.getConfiguration());
      if (!fs.exists(rootDir)) {
        fs.mkdirs(rootDir);
      }
      Path tablePath;
      if (CatalogUtil.isFQTableName(tableName)) {
        Pair<String, String> name = CatalogUtil.separateQualifierAndName(tableName);
        tablePath = new Path(rootDir, new Path(name.getFirst(), name.getSecond()));
      } else {
        tablePath = new Path(rootDir, tableName);
      }

      fs.mkdirs(tablePath);
      if (tableDatas.length > 0) {
        int recordPerFile = tableDatas.length / numDataFiles;
        if (recordPerFile == 0) {
          recordPerFile = 1;
        }
        FSDataOutputStream out = null;
        for (int j = 0; j < tableDatas.length; j++) {
          if (out == null || j % recordPerFile == 0) {
            if (out != null) {
              out.close();
            }
            Path dfsPath = new Path(tablePath, tableName + j + ".tbl");
            out = fs.create(dfsPath);
          }
          out.write((tableDatas[j] + "\n").getBytes());
        }
        if (out != null) {
          out.close();
        }
      }
      TableMeta meta = CatalogUtil.newTableMeta("TEXT", tableOption);
      client.createExternalTable(tableName, schema, tablePath.toUri(), meta);
    } finally {
      client.close();
    }
  }

    /**
    * Write lines to a file.
    *
    * @param file File to write lines to
    * @param lines Strings written to the file
    * @throws java.io.IOException
    */
  private static void writeLines(File file, String... lines)
      throws IOException {
    Writer writer = Files.newWriter(file, Charsets.UTF_8);
    try {
      for (String line : lines) {
        writer.write(line);
        writer.write('\n');
      }
    } finally {
      Closeables.closeQuietly(writer);
    }
  }

  public void setAllTajoDaemonConfValue(String key, String value) {
    tajoMaster.getContext().getConf().set(key, value);
    setAllWorkersConfValue(key, value);
  }

  public void setAllWorkersConfValue(String key, String value) {
    for (TajoWorker eachWorker: tajoWorkers) {
      eachWorker.getConfig().set(key, value);
    }
  }

  public void waitForQuerySubmitted(QueryId queryId) throws Exception {
    waitForQuerySubmitted(queryId, 50);
  }

  public void waitForQuerySubmitted(QueryId queryId, int delay) throws Exception {
    QueryMasterTask qmt = null;

    int i = 0;
    while (qmt == null || TajoClientUtil.isQueryWaitingForSchedule(qmt.getState())) {
      try {
        Thread.sleep(delay);

        if (qmt == null) {
          qmt = getQueryMasterTask(queryId);
        }
      } catch (InterruptedException e) {
      }
      if (++i > 200) {
        throw new IOException("Timed out waiting for query to start");
      }
    }
  }

  public void waitForQueryState(Query query, TajoProtos.QueryState expected, int delay) throws Exception {
    int i = 0;
    while (query == null || query.getSynchronizedState() != expected) {
      try {
        Thread.sleep(delay);
      } catch (InterruptedException e) {
      }
      if (++i > 200) {
        throw new IOException("Timed out waiting. expected: " + expected +
            ", actual: " + query != null ? String.valueOf(query.getSynchronizedState()) : String.valueOf(query));
      }
    }
  }

  public void waitForStageState(Stage stage, StageState expected, int delay) throws Exception {

    int i = 0;
    while (stage == null || stage.getSynchronizedState() != expected) {
      try {
        Thread.sleep(delay);
      } catch (InterruptedException e) {
      }
      if (++i > 200) {
        throw new IOException("Timed out waiting");
      }
    }
  }

  public QueryMasterTask getQueryMasterTask(QueryId queryId) {
    QueryMasterTask qmt = null;
    for (TajoWorker worker : getTajoWorkers()) {
      qmt = worker.getWorkerContext().getQueryMaster().getQueryMasterTask(queryId, true);
      if (qmt != null && queryId.equals(qmt.getQueryId())) {
        break;
      }
    }
    return qmt;
  }

  public QueryHistory getQueryHistory(QueryId queryId) throws IOException {
    QueryHistory queryHistory = null;
    for (TajoWorker worker : getTajoWorkers()) {
      queryHistory = worker.getWorkerContext().getQueryMaster().getQueryHistory(queryId);
      if (queryHistory != null) {
        break;
      }
    }
    return queryHistory;
  }
}