package com.waitingforcode;

import org.openjdk.jmh.annotations.*;

import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@OutputTimeUnit(TimeUnit.MILLISECONDS)
@BenchmarkMode(Mode.All)
public class ReadMicroBenchmark {

    @State(Scope.Thread)
    public static class InitializationState {

        private CassandraSession cassandraSession = new CassandraSession();
        {
            String colsToInsertInto80ColsTable = getColumnsToInsertClause(cassandraSession.getWideRowTable80Columns());
            String valuesToInsert80ColsTable = getValuesToInsertClause(cassandraSession.getWideRowTable80Columns());
            String colsToInsertInto40ColsTable = getColumnsToInsertClause(cassandraSession.getWideRowTable40Columns());
            String valuesToInsert40ColsTable = getValuesToInsertClause(cassandraSession.getWideRowTable40Columns());
            String colsToInsertInto20ColsTable = getColumnsToInsertClause(cassandraSession.getWideRowTable20Columns());
            String valuesToInsert20ColsTable = getValuesToInsertClause(cassandraSession.getWideRowTable20Columns());
            cassandraSession.createTables();
            for (int i = 0; i < ROWS_TO_INSERT; i++) {
                cassandraSession.execute("INSERT INTO wide_rows_test_80 (id, "+colsToInsertInto80ColsTable+") VALUES (" +
                        i+", "+valuesToInsert80ColsTable+")");
                cassandraSession.execute("INSERT INTO wide_rows_test_40 (id, "+colsToInsertInto40ColsTable+") VALUES (" +
                        i+", "+valuesToInsert40ColsTable+")");
                cassandraSession.execute("INSERT INTO wide_rows_test_20 (id, "+colsToInsertInto20ColsTable+") VALUES (" +
                        i+", "+valuesToInsert20ColsTable+")");
                for (int j = 0; j < cassandraSession.getWideRowTable80Columns().size(); j++) {
                    String key = i+"_"+cassandraSession.getWideRowTable80Columns().get(j);
                    cassandraSession.execute("INSERT INTO skinny_row_test (col, value) VALUES ('"+key+"', 'value"+i+"')");
                }
            }
        }

        private static final int ROWS_TO_INSERT = 100;

        private static String getColumnsToInsertClause(List<String> columns) {
            return columns.stream().collect(Collectors.joining(","));
        }

        private static String getValuesToInsertClause(List<String> columns) {
            return columns.stream()
                    .map(colName -> "'"+colName + " value'")
                    .collect(Collectors.joining(","));
        }

        @Setup(Level.Trial)
        public void doSetup() {
        }

        @TearDown(Level.Trial)
        public void doTearDown() {
            cassandraSession.closeSession();
        }
    }

    @Benchmark
    public void read_all_wide_rows_80_cols(InitializationState initializationState) {
        for (int i = 0; i < initializationState.ROWS_TO_INSERT; i++) {
            initializationState.cassandraSession.execute("SELECT * FROM wide_rows_test_80 WHERE id = "+i);
        }
    }

    @Benchmark
    public void read_all_wide_rows_40_cols(InitializationState initializationState) {
        for (int i = 0; i < initializationState.ROWS_TO_INSERT; i++) {
            initializationState.cassandraSession.execute("SELECT * FROM wide_rows_test_40 WHERE id = "+i);
        }
    }

    @Benchmark
    public void read_all_wide_rows_20_cols(InitializationState initializationState) {
        for (int i = 0; i < initializationState.ROWS_TO_INSERT; i++) {
            initializationState.cassandraSession.execute("SELECT * FROM wide_rows_test_20 WHERE id = "+i);
        }
    }

    @Benchmark
    public void read_all_skinny_rows(InitializationState initializationState) {
        for (int i = 0; i < initializationState.ROWS_TO_INSERT; i++) {
            for (int j = 0; j < initializationState.cassandraSession.getWideRowTable80Columns().size(); j++) {
                String key = i+"_"+initializationState.cassandraSession.getWideRowTable80Columns().get(j);
                initializationState.cassandraSession.execute("SELECT * FROM skinny_row_test WHERE col = '"+key+"'");
            }
        }
    }

    @Benchmark
    public void read_20_columns_from_all_wide_row_80_cols_table(InitializationState initializationState) {
        for (int i = 0; i < initializationState.ROWS_TO_INSERT; i++) {
            initializationState.cassandraSession.execute("SELECT id, column0, column1, column2, column3, column4, column5, column6, column7, column8," +
                    "column9, column10, column11, column12, column13, column14, column15, column16, column17, " +
                    "column18, column19 FROM wide_rows_test_80 WHERE id = "+i);
        }
    }

    @Benchmark
    public void read_20_columns_from_all_wide_row_40_cols_table(InitializationState initializationState) {
        for (int i = 0; i < initializationState.ROWS_TO_INSERT; i++) {
            initializationState.cassandraSession.execute("SELECT id, column0, column1, column2, column3, column4, column5, column6, column7, column8," +
                    "column9, column10, column11, column12, column13, column14, column15, column16, column17, " +
                    "column18, column19 FROM wide_rows_test_40 WHERE id = "+i);
        }
    }

    @Benchmark
    public void read_20_columns_from_all_wide_row_20_cols_table(InitializationState initializationState) {
        for (int i = 0; i < initializationState.ROWS_TO_INSERT; i++) {
            initializationState.cassandraSession.execute("SELECT id, column0, column1, column2, column3, column4, column5, column6, column7, column8," +
                    "column9, column10, column11, column12, column13, column14, column15, column16, column17, " +
                    "column18, column19 FROM wide_rows_test_20 WHERE id = "+i);
        }
    }

    @Benchmark
    public void read_20_rows_from_all_skinny_rows(InitializationState initializationState) {
        for (int i = 0; i < initializationState.ROWS_TO_INSERT; i++) {
            for (int j = 0; j < 20; j++) {
                String key = i+"_"+initializationState.cassandraSession.getWideRowTable80Columns().get(j);
                initializationState.cassandraSession.execute("SELECT * FROM skinny_row_test  WHERE col = '"+key+"'");
            }
        }
    }

}