/*
 * Copyright 2021 4Paradigm
 *
 * Licensed 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 com._4paradigm.openmldb.java_sdk_test.performance;

import com._4paradigm.openmldb.sdk.SdkOption;
import com._4paradigm.openmldb.sdk.SqlExecutor;
import com._4paradigm.openmldb.sdk.impl.SqlClusterExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.*;
import java.util.Random;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;

public class FESQLMixerWorkloadExample extends BaseExample {
    private static final Logger logger = LoggerFactory.getLogger(RequestAggExample.class);
    private SqlExecutor executor;
    private SdkOption option;
    private String db = "dbbian";
    private String cardDdl = "create table card_info(\n" +
            "crd_nbr string,\n" +
            "clt_nbr string,\n" +
            "act_nbr string,\n" +
            "crd_atv_flg int,\n" +
            "cyc_day int,\n" +
            "crd_hdr_typ int,\n" +
            "chp_crd_flg int,\n" +
            "crd_pur_pin_flg int,\n" +
            "crd_csh_pin_flg int,\n" +
            "crd_exp_flg int,\n" +
            "crd_vip_drp int,\n" +
            "crd_isu_dte timestamp,\n" +
            "crd_exp_dte timestamp,\n" +
            "crd_lst_isu_dte timestamp,\n" +
            "crd_atv_dte timestamp,\n" +
            "index(key=crd_nbr, ts=crd_lst_isu_dte, ttl=2, ttl_type=latest));";


    private String tranDdl1 = "create table tran(\n" +
            "tran_no string,\n" +
            "card_no string,\n" +
            "trx_date date,\n" +
            "trx_time timestamp,\n" +
            "trx_amt double,\n" +
            "input_source int,\n" +
            "pos_entry_mode int,\n" +
            "acq_id int,\n" +
            "merchant_type int,\n" +
            "terminal int,\n" +
            "merchant_id int,\n" +
            "country_code int,\n" +
            "crdh_pre_ind int,\n" +
            "usd_amt double,\n" +
            "org_amt double,\n" +
            "org_curr_code int,\n" +
            "term_city int,\n" +
            "term_st int,\n" +
            "pin_check int,\n" +
            "tran_label int,\n" +
            "index(key=card_no, ts=trx_time, ttl=1d, ttl_type=absolute));";
    private boolean setupOk = false;
    private int recordSize = 10000;
    private String cardFormat = "insert into card_info values('%s', 'clt_nbr0'," +
            "'act_nbr0', 1, 2, 3, 4, 5, 6, 7, 8, 1596697833000l, 1596697833000l, %dl, 1596697833000l);";
    private String tranFormat = "insert into tran values('tran_noxx', '%s', '2020-10-10', %dl, 1.0, 1, 2,3,4,5,6,7,8, 1.0, 2.0, 1, 2, 3, 4,5);";
    private Random random = new Random(System.currentTimeMillis());
    private ScheduledExecutorService sched = Executors.newScheduledThreadPool(1);
    private long counter = 0;
    private String cardId = "cardId";

    public FESQLMixerWorkloadExample() {
    }

    public void print() {
        logger.info("ddl:\n{}", cardDdl);
        logger.info("ddl:\n{}", tranDdl1);
        logger.info("insert:\n{}", String.format(cardFormat, cardId, System.currentTimeMillis()));

        String key = String.valueOf(random.nextInt(100000));
        logger.info("insert:\n{}", String.format(tranFormat, key, System.currentTimeMillis()));
    }

    public void initExecutor() {
        SdkOption sdkOption = new SdkOption();
        sdkOption.setSessionTimeout(30000);
        sdkOption.setZkCluster(zkCluster);
        sdkOption.setZkPath(zkPath);
        sdkOption.setEnableDebug(true);
        this.option = sdkOption;
        try {
            executor = new SqlClusterExecutor(option);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void setup() {
        executor.createDB(db);
        executor.executeDDL(db, cardDdl);
        setupOk = executor.executeDDL(db, tranDdl1);
        makeData();
       /* sched.scheduleWithFixedDelay(new Runnable() {
            public void run() {
                makeData();
            }
        }, 0, 1, TimeUnit.MINUTES);*/
    }

    public void makeData() {
        String sql = String.format(cardFormat, cardId, System.currentTimeMillis());
        executor.executeInsert(db, sql);
        for (int i = 0; i < 100; i++) {
            sql = String.format(tranFormat, cardId, System.currentTimeMillis() - i);
            boolean ok = executor.executeInsert(db, sql);
            System.out.println(ok);
        }
        String key = String.valueOf(random.nextInt(100000));
        executor.executeInsert(db, sql);
        for (int i = 0; i < 1000; i++) {
            sql = String.format(tranFormat, key, System.currentTimeMillis() - i);
            executor.executeInsert(db, sql);
        }
    }

//    public void lastJoinBm() {
//        String sql = "select * from\n" +
//                "(select\n" +
//                "card_no,\n" +
//                "trx_time,\n" +
//                "merchant_id,\n" +
////                "month(trx_time) as fea_month,\n" +
////                "dayofmonth(trx_time) as fea_month,\n" +
////                "hour(trx_time) as fea_hour,\n" +
////                "week(trx_time) as fea_week,\n" +
////                "substr(card_no, 1, 6) as card_no_prefix,\n" +
////                "max(trx_amt) over w30d as w30d_trx_max ,\n" +
////                "min(trx_amt) over w30d as w30d_trx_min,\n" +
////                "sum(trx_amt) over w30d,\n" +
////                "avg(trx_amt) over w30d,\n" +
////                "max(usd_amt) over w30d,\n" +
////                "min(usd_amt) over w30d,\n" +
////                "sum(usd_amt) over w30d,\n" +
////                "avg(usd_amt) over w30d,\n" +
////                "max(org_amt) over w30d,\n" +
////                "min(org_amt) over w30d,\n" +
////                "sum(org_amt) over w30d,\n" +
////                "avg(org_amt) over w30d,\n" +
////                "distinct_count(merchant_id) over w30d,\n" +
////                "count(merchant_id) over w30d,\n" +
////                "distinct_count(term_city) over w30d,\n" +
////                "count(term_city) over w30d,\n" +
////                "max(trx_amt) over w10d,\n" +
////                "min(trx_amt) over w10d,\n" +
////                "sum(trx_amt) over w10d,\n" +
////                "avg(trx_amt) over w10d,\n" +
////                "max(usd_amt) over w10d,\n" +
////                "min(usd_amt) over w10d,\n" +
////                "sum(usd_amt) over w10d,\n" +
////                "avg(usd_amt) over w10d,\n" +
////                "max(org_amt) over w10d,\n" +
////                "min(org_amt) over w10d,\n" +
////                "sum(org_amt) over w10d,\n" +
////                "avg(org_amt) over w10d,\n" +
//                "distinct_count(merchant_id) over w10d,\n" +
//                "count(merchant_id)  over w10d,\n" +
//                "distinct_count(term_city)  over w10d,\n" +
//                "count(term_city) over w10d\n" +
//                "from  tran\n" +
//                "window w30d as (PARTITION BY tran.card_no ORDER BY tran.trx_time ROWS_RANGE BETWEEN 30d PRECEDING AND CURRENT ROW),\n" +
//                "w10d as (PARTITION BY tran.card_no ORDER BY tran.trx_time ROWS_RANGE BETWEEN 10d PRECEDING AND CURRENT ROW)) as trx_fe\n" +
//                "last join card_info order by card_info.crd_lst_isu_dte on trx_fe.card_no = card_info.crd_nbr and trx_fe.trx_time >= card_info.crd_lst_isu_dte;\n";
//        String key = cardId;
//        String trans = "trans";
//        SQLRequestRow row = executor.getRequestRow(db, sql);
//        row.Init(key.length());
//        row.AppendNULL();
//        row.AppendString(key);
//        row.AppendNULL();
//        row.AppendTimestamp(System.currentTimeMillis());
//        row.AppendDouble(10.0);
//        row.AppendInt32(1);
//        row.AppendInt32(2);
//        row.AppendInt32(3);
//        row.AppendInt32(4);
//        row.AppendInt32(5);
//        row.AppendInt32(6);
//        row.AppendInt32(7);
//        row.AppendInt32(8);
//        row.AppendDouble(1.0);
//        row.AppendDouble(2.0);
//        row.AppendInt32(1);
//        row.AppendInt32(2);
//        row.AppendInt32(3);
//        row.AppendInt32(4);
//        row.AppendInt32(5);
//        row.Build();
//        ResultSet rs = executor.executeSQL(db, sql, row);
//        row.delete();
//        if (rs == null) {
//            System.out.println("no result");
//        } else {
//            System.out.println("ok");
//        }
//        rs.Next();
//        Schema schema = rs.GetSchema();
//        for (int i = 0; i < schema.GetColumnCnt(); i++) {
//            System.out.println(schema.GetColumnName(i));
//            if (rs.IsNULL(i)) {
//                System.out.println("null");
//                continue;
//            }
//            DataType type = schema.GetColumnType(i);
//            if (type.swigValue() == DataType.kTypeInt16.swigValue()) {
//                System.out.println(rs.GetInt16Unsafe(i));
//            } else if (type.swigValue() == DataType.kTypeInt32.swigValue()) {
//                System.out.println(rs.GetInt32Unsafe(i));
//            } else if (type.swigValue() == DataType.kTypeInt64.swigValue()) {
//                System.out.println(rs.GetInt64Unsafe(i));
//            } else if (type.swigValue() == DataType.kTypeFloat.swigValue()) {
//                System.out.println(rs.GetFloatUnsafe(i));
//            } else if (type.swigValue() == DataType.kTypeDouble.swigValue()) {
//                System.out.println(rs.GetDoubleUnsafe(i));
//            } else if (type.swigValue() == DataType.kTypeString.swigValue()) {
//                System.out.println(rs.GetStringUnsafe(i));
//            } else if (type.swigValue() == DataType.kTypeTimestamp.swigValue()) {
//                System.out.println(rs.GetTimeUnsafe(i));
//            }
//        }
//        FesqlUtil.show(rs);
//        schema.delete();
//        rs.delete();
//    }

    public void lastJoinBm() {
        String sql = "select * from\n" +
                "(select\n" +
                "card_no,\n" +
                "trx_time,\n" +
                "merchant_id,\n" +
//                "month(trx_time) as fea_month,\n" +
//                "dayofmonth(trx_time) as fea_month,\n" +
//                "hour(trx_time) as fea_hour,\n" +
//                "week(trx_time) as fea_week,\n" +
//                "substr(card_no, 1, 6) as card_no_prefix,\n" +
//                "max(trx_amt) over w30d as w30d_trx_max ,\n" +
//                "min(trx_amt) over w30d as w30d_trx_min,\n" +
//                "sum(trx_amt) over w30d,\n" +
//                "avg(trx_amt) over w30d,\n" +
//                "max(usd_amt) over w30d,\n" +
//                "min(usd_amt) over w30d,\n" +
//                "sum(usd_amt) over w30d,\n" +
//                "avg(usd_amt) over w30d,\n" +
//                "max(org_amt) over w30d,\n" +
//                "min(org_amt) over w30d,\n" +
//                "sum(org_amt) over w30d,\n" +
//                "avg(org_amt) over w30d,\n" +
//                "distinct_count(merchant_id) over w30d,\n" +
//                "count(merchant_id) over w30d,\n" +
//                "distinct_count(term_city) over w30d,\n" +
//                "count(term_city) over w30d,\n" +
//                "max(trx_amt) over w10d,\n" +
//                "min(trx_amt) over w10d,\n" +
//                "sum(trx_amt) over w10d,\n" +
//                "avg(trx_amt) over w10d,\n" +
//                "max(usd_amt) over w10d,\n" +
//                "min(usd_amt) over w10d,\n" +
//                "sum(usd_amt) over w10d,\n" +
//                "avg(usd_amt) over w10d,\n" +
//                "max(org_amt) over w10d,\n" +
//                "min(org_amt) over w10d,\n" +
//                "sum(org_amt) over w10d,\n" +
//                "avg(org_amt) over w10d,\n" +
                "distinct_count(merchant_id) over w10d,\n" +
                "count(merchant_id)  over w10d,\n" +
                "distinct_count(term_city)  over w10d,\n" +
                "count(term_city) over w10d\n" +
                "from  tran\n" +
                "window w30d as (PARTITION BY tran.card_no ORDER BY tran.trx_time ROWS_RANGE BETWEEN 30d PRECEDING AND CURRENT ROW),\n" +
                "w10d as (PARTITION BY tran.card_no ORDER BY tran.trx_time ROWS_RANGE BETWEEN 10d PRECEDING AND CURRENT ROW)) as trx_fe\n" +
                "last join card_info order by card_info.crd_lst_isu_dte on trx_fe.card_no = card_info.crd_nbr and trx_fe.trx_time >= card_info.crd_lst_isu_dte;\n";
        String key = cardId;
        String trans = "trans";
        PreparedStatement pst = null;
        ResultSet resultSet = null;
        try {
            pst = executor.getRequestPreparedStmt(db, sql);
            pst.setNull(1, 0);
            pst.setString(2, key);
            pst.setNull(3, 0);
            pst.setTimestamp(4, new Timestamp(System.currentTimeMillis()));
            pst.setDouble(5, 10.0);
            pst.setInt(6, 1);
            pst.setInt(7, 2);
            pst.setInt(8, 3);
            pst.setInt(9, 4);
            pst.setInt(10, 5);
            pst.setInt(11, 6);
            pst.setInt(12, 7);
            pst.setInt(13, 8);
            pst.setDouble(14, 1.0);
            pst.setDouble(15, 2.0);
            pst.setInt(16, 1);
            pst.setInt(17, 2);
            pst.setInt(18, 3);
            pst.setInt(19, 4);
            pst.setInt(20, 5);
            resultSet = pst.executeQuery();
            if (resultSet == null) {
                System.out.println("no result");
            } else {
                System.out.println("ok");
            }
            boolean next = resultSet.next();
            if (!next) {
                System.out.println("next failed");
            }
            ResultSetMetaData metaData = resultSet.getMetaData();
            for (int i = 0; i < metaData.getColumnCount(); i++) {
                System.out.println(metaData.getColumnName(i + 1));
                if (resultSet.getNString(i) == null) {
                    System.out.println("null");
                    continue;
                }
                int columnType = metaData.getColumnType(i + 1);
                if (columnType == Types.SMALLINT) {
                    System.out.println(resultSet.getShort(i + 1));
                } else if (columnType == Types.INTEGER) {
                    System.out.println(resultSet.getInt(i + 1));
                } else if (columnType == Types.BIGINT) {
                    System.out.println(resultSet.getLong(i + 1));
                } else if (columnType == Types.FLOAT) {
                    System.out.println(resultSet.getFloat(i + 1));
                } else if (columnType == Types.DOUBLE) {
                    System.out.println(resultSet.getDouble(i + 1));
                } else if (columnType == Types.VARCHAR) {
                    System.out.println(resultSet.getString(i + 1));
                } else if (columnType == Types.TIMESTAMP) {
                    System.out.println(resultSet.getTimestamp(i + 1).getTime());
                }
            }
        } catch (SQLException throwables) {
            throwables.printStackTrace();
        } finally {
            try {
                if (pst != null) {
                    pst.close();
                    if (resultSet != null) {
                        resultSet.close();
                    }
                }
            } catch (SQLException throwables) {
                throwables.printStackTrace();
            }
        }
    }

    public static void run() {
        final FESQLMixerWorkloadExample example = new FESQLMixerWorkloadExample();
        example.initExecutor();
        example.setup();
        example.lastJoinBm();
    }


}
