<commit_before>package com.splicemachine.perf.runner;

import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Lists;
import com.google.common.collect.Multimap;
import com.splicemachine.derby.stats.Accumulator;
import com.splicemachine.derby.stats.Stats;
import com.splicemachine.derby.stats.TimeUtils;
import com.splicemachine.derby.stats.TimingStats;
import com.splicemachine.perf.runner.qualifiers.Result;
import com.splicemachine.tools.ConnectionPool;
import com.splicemachine.utils.SpliceLogUtils;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.log4j.Logger;

import java.io.PrintStream;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

/**
 * @author <NAME>
 *Created on: 3/15/13
 */
public class Table {
    private static final Logger LOG = Logger.getLogger(Table.class);
    private final String name;
    private final List<Column> columns;
    private final int numRows;
    private final int insertBatch;
    private final int insertThreads;
    private final List<String> files;

    public Table(String name, List<Column> columns, int numRows, int insertBatch, int insertThreads, List<String> filesToImport) {
        this.name = name;
        this.columns = columns;
        this.numRows = numRows;
        this.insertBatch = insertBatch;
        this.insertThreads = insertThreads;
        this.files = filesToImport;
    }

    public void create(ConnectionPool pool) throws SQLException, InterruptedException {
        SpliceLogUtils.debug(LOG, "Creating table %s",name);
        Connection conn = null;
        try{
            conn = pool.acquire();
            String createStr = getCreateTableString();
            SpliceLogUtils.trace(LOG,"using sql "+ createStr);
            PreparedStatement ps = conn.prepareStatement(createStr);
            ps.execute();
        }finally{
            if(conn!=null)
                conn.close();
        }
    }

    public Result insertData(final ConnectionPool connectionPool) throws Exception{
        if(files !=null){
            return importData(connectionPool);
        }else{
            return generateAndInsertData(connectionPool);
        }
    }

    private Result importData(ConnectionPool connectionPool) throws InterruptedException, SQLException {
        ImportResult stats = new ImportResult();
        long start = System.nanoTime();
        for(String file: files){
            Connection connection = connectionPool.acquire();
            try{
                SpliceLogUtils.debug(LOG, "Loading data from file %s", file);
                long fileStart = System.nanoTime();
                PreparedStatement s = connection.prepareStatement("call SYSCS_UTIL.SYSCS_IMPORT_DATA(null,'"+name.toUpperCase()+"',null,null,'"+file+"',',',null,null)");
                s.execute();
                long stop = System.nanoTime();
                stats.addTime(file,stop-fileStart);
            }finally{
                connection.close();
            }
        }
        stats.finalize(System.nanoTime()-start);
        return stats;
    }

    private Result generateAndInsertData(final ConnectionPool connectionPool) throws InterruptedException, ExecutionException {
        ExecutorService dataInserter = Executors.newFixedThreadPool(insertThreads);
        try{
            List<Future<Void>> futures = Lists.newArrayListWithCapacity(insertThreads + 1);
            final Accumulator insertAccumulator = TimingStats.uniformSafeAccumulator();
            insertAccumulator.start();
            for(int i=0;i<insertThreads;i++){
                futures.add(dataInserter.submit(new Callable<Void>() {
                    @Override
                    public Void call() throws Exception {
                        int rowsToInsert = numRows/insertThreads;
                        int tenPercentSize = rowsToInsert>10?rowsToInsert/10 : 1;
                        int written = 0;
                        Connection conn = connectionPool.acquire();
                        try{
                            PreparedStatement ps = conn.prepareStatement(getInsertDataString());
                            for(int numRowsInserted=0;numRowsInserted<numRows/insertThreads;numRowsInserted++){
                                fillStatement(ps);
                                ps.addBatch();
                                written++;
                                if(numRowsInserted>0&&numRowsInserted%insertBatch==0){
                                    long start = System.nanoTime();
                                    int numInserted = ps.executeBatch().length;
                                    insertAccumulator.tick(numInserted,System.nanoTime()-start);

                                    if((written-1)%tenPercentSize==0)
                                        SpliceLogUtils.info(LOG, "%d%% complete", ((written - 1) / tenPercentSize) * 10);
                                }
                            }
                            long start = System.nanoTime();
                            int numInserted = ps.executeBatch().length;
                            if(numInserted>0){
                                insertAccumulator.tick(numInserted, System.nanoTime() - start);
                                SpliceLogUtils.info(LOG,"writing complete");
                            }<commit_msg>TestRunner - start a transaction in each inserter thread<commit_after>package com.splicemachine.perf.runner;

import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Lists;
import com.google.common.collect.Multimap;
import com.splicemachine.derby.stats.Accumulator;
import com.splicemachine.derby.stats.Stats;
import com.splicemachine.derby.stats.TimeUtils;
import com.splicemachine.derby.stats.TimingStats;
import com.splicemachine.perf.runner.qualifiers.Result;
import com.splicemachine.tools.ConnectionPool;
import com.splicemachine.utils.SpliceLogUtils;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.log4j.Logger;

import java.io.PrintStream;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

/**
 * @author <NAME>
 *Created on: 3/15/13
 */
public class Table {
    private static final Logger LOG = Logger.getLogger(Table.class);
    private final String name;
    private final List<Column> columns;
    private final int numRows;
    private final int insertBatch;
    private final int insertThreads;
    private final List<String> files;

    public Table(String name, List<Column> columns, int numRows, int insertBatch, int insertThreads, List<String> filesToImport) {
        this.name = name;
        this.columns = columns;
        this.numRows = numRows;
        this.insertBatch = insertBatch;
        this.insertThreads = insertThreads;
        this.files = filesToImport;
    }

    public void create(ConnectionPool pool) throws SQLException, InterruptedException {
        SpliceLogUtils.debug(LOG, "Creating table %s",name);
        Connection conn = null;
        try{
            conn = pool.acquire();
            String createStr = getCreateTableString();
            SpliceLogUtils.trace(LOG,"using sql "+ createStr);
            PreparedStatement ps = conn.prepareStatement(createStr);
            ps.execute();
        }finally{
            if(conn!=null)
                conn.close();
        }
    }

    public Result insertData(final ConnectionPool connectionPool) throws Exception{
        if(files !=null){
            return importData(connectionPool);
        }else{
            return generateAndInsertData(connectionPool);
        }
    }

    private Result importData(ConnectionPool connectionPool) throws InterruptedException, SQLException {
        ImportResult stats = new ImportResult();
        long start = System.nanoTime();
        for(String file: files){
            Connection connection = connectionPool.acquire();
            try{
                SpliceLogUtils.debug(LOG, "Loading data from file %s", file);
                long fileStart = System.nanoTime();
                PreparedStatement s = connection.prepareStatement("call SYSCS_UTIL.SYSCS_IMPORT_DATA(null,'"+name.toUpperCase()+"',null,null,'"+file+"',',',null,null)");
                s.execute();
                long stop = System.nanoTime();
                stats.addTime(file,stop-fileStart);
            }finally{
                connection.close();
            }
        }
        stats.finalize(System.nanoTime()-start);
        return stats;
    }

    private Result generateAndInsertData(final ConnectionPool connectionPool) throws InterruptedException, ExecutionException {
        ExecutorService dataInserter = Executors.newFixedThreadPool(insertThreads);
        try{
            List<Future<Void>> futures = Lists.newArrayListWithCapacity(insertThreads + 1);
            final Accumulator insertAccumulator = TimingStats.uniformSafeAccumulator();
            insertAccumulator.start();
            for(int i=0;i<insertThreads;i++){
                futures.add(dataInserter.submit(new Callable<Void>() {
                    @Override
                    public Void call() throws Exception {
                        int rowsToInsert = numRows/insertThreads;
                        int tenPercentSize = rowsToInsert>10?rowsToInsert/10 : 1;
                        int written = 0;
                        Connection conn = connectionPool.acquire();
                        conn.setAutoCommit(false);
                        try{
                            PreparedStatement ps = conn.prepareStatement(getInsertDataString());
                            for(int numRowsInserted=0;numRowsInserted<numRows/insertThreads;numRowsInserted++){
                                fillStatement(ps);
                                ps.addBatch();
                                written++;
                                if(numRowsInserted>0&&numRowsInserted%insertBatch==0){
                                    long start = System.nanoTime();
                                    int numInserted = ps.executeBatch().length;
                                    insertAccumulator.tick(numInserted,System.nanoTime()-start);

                                    if((written-1)%tenPercentSize==0)
                                        SpliceLogUtils.info(LOG, "%d%% complete", ((written - 1) / tenPercentSize) * 10);
                                }
                            }
                            long start = System.nanoTime();
                            int numInserted = ps.executeBatch().length;
                            if(numInserted>0){
                                insertAccumulator.tick(numInserted, System.nanoTime() - start);
                                SpliceLogUtils.info(LOG,"writing complete");
                            }
{'fraction_non_alphanumeric': 0.0935857945279942, 'fraction_numerical': 0.0041674216343540495, 'mean_word_length': 2.016120218579235, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '140664', 'n_tokens_mistral': 2859, 'n_tokens_neox': 2759, 'n_words': 558}