package mayton.watchdog.pg

import java.io.{FileOutputStream, PrintWriter}
import java.sql.Connection

import mayton.watchdog.pg.Utils._

import scala.collection.mutable

object PgWatchdogEAV {

  def EAV_LOG = "eav_log"
  def TRIGGER_SUFFIX = "_trigger"
  def FUNC_SUFFIX = "_func"

  def createDefaultEAVLog(script : PrintWriter) : Unit = {
    script.println(
      s"""|CREATE TABLE $EAV_LOG (
          |  TS          TIMESTAMP   NOT NULL,
          |  OPERATION   CHAR(1)     NOT NULL CHECK (OPERATION IN ('I','U','D')),
          |  TABLE_NAME  VARCHAR(64) NOT NULL,
          |  COLUMN_NAME VARCHAR(64) NOT NULL,
          |  COL_VALUE   TEXT);
          |
          |  """.stripMargin)
  }

  def eavTriggersFunctionScript(connection: Connection, script: PrintWriter) : Unit = {
    for(tableName <- tables(connection).filter(p => p != EAV_LOG)) {
      val columnDefinitions : List[ColumnDefinition] = getColumnNames(connection, tableName)
      eavFunctionScript(script, tableName, columnDefinitions)
    }
  }

  def eavFunctionScript(script : PrintWriter, tableName : String, columnDefinitions : List[ColumnDefinition]) : Unit = {

    script.print(
      s"""CREATE OR REPLACE FUNCTION $tableName$FUNC_SUFFIX() RETURNS TRIGGER AS $$$$
         |BEGIN
         |  IF TG_OP = 'INSERT' THEN
         |""".stripMargin)

    for(columnDefinition <- columnDefinitions) {
      script.print(s"    IF NEW.${columnDefinition.columnName} IS NOT NULL THEN\n")
      script.print(s"      INSERT INTO $EAV_LOG(TS, OPERATION, TABLE_NAME, COLUMN_NAME, COL_VALUE) VALUES(CURRENT_TIMESTAMP, 'I', '$tableName', '${columnDefinition.columnName}', NEW.${columnDefinition.columnName});\n")
      script.print(s"    END IF;\n")
      script.print(s"    RETURN NEW;\n")
    }

    script.printf("  ELSIF TG_OP = 'UPDATE' THEN\n")
    for(columnDefinition <- columnDefinitions) {
      script.print(s"    IF NEW.${columnDefinition.columnName} IS NOT NULL THEN\n")
      script.print(s"      INSERT INTO $EAV_LOG(TS, OPERATION, TABLE_NAME, COLUMN_NAME, COL_VALUE) VALUES(CURRENT_TIMESTAMP, 'U', '$tableName', '${columnDefinition.columnName}', NEW.${columnDefinition.columnName});\n")
      script.print(s"    END IF;\n")
      script.print(s"    RETURN NEW;\n")
    }

    script.printf("  ELSIF TG_OP = 'DELETE' THEN\n")
    for(columnDefinition <- columnDefinitions) {
      script.print(s"    IF OLD.${columnDefinition.columnName} IS NOT NULL THEN\n")
      script.print(s"      INSERT INTO $EAV_LOG(TS, OPERATION, TABLE_NAME, COLUMN_NAME, COL_VALUE) VALUES(CURRENT_TIMESTAMP, 'D', '$tableName', '${columnDefinition.columnName}', OLD.${columnDefinition.columnName});\n")
      script.print(s"    END IF;\n")
      script.print(s"    RETURN OLD;\n")
    }

    script.print(
     s"""|  END IF;
         |--RETURN NEW;
         |END;
         |$$$$
         |LANGUAGE plpgsql;
         |
         |""".stripMargin)

    script.print(
      s"""
         |CREATE TRIGGER $tableName$TRIGGER_SUFFIX AFTER UPDATE OR INSERT OR DELETE
         |   ON $tableName
         |   FOR EACH ROW EXECUTE PROCEDURE $tableName$FUNC_SUFFIX();
         |
         |""".stripMargin)
  }


  def process(connection: Connection, script : PrintWriter) : Boolean = {
    createDefaultEAVLog(script)
    eavTriggersFunctionScript(connection, script)
    true
  }

  def main(arg : Array[String]) : Unit = {
    val props: mutable.Map[String, String] = toScalaMutableMap(tryToLoadSensitiveProperties())
    val host = props("host")
    val port = props("port")
    val database = props("database")
    val connection = createConnection(s"jdbc:postgresql://$host:$port/$database", props("user"), props("password"))
    val script = new PrintWriter(new FileOutputStream("out/pg-watchdog-tables-eav.sql"))
    process(connection, script)
    script.close()
  }

}

package ha.jdbc_test;

public class HAJDBC_Test {

    private static final java.util.ArrayList<String> URLS = new java.util.ArrayList<String>() {
        {
            add("jdbc:h2:./test_h2;MODE=PostgreSQL;DATABASE_TO_LOWER=TRUE;USER=sa;PASSWORD=");
            add("jdbc:postgresql://127.0.0.1/test_pg?user=postgres&password=secret");
        }
    };

    public HAJDBC_Test() {
    }

    public static void main(String[] args) throws Exception {

        long t = System.currentTimeMillis();

        HAJDBC_Test ha = new HAJDBC_Test();

        for (String url : URLS) {
            try (java.sql.Connection c = java.sql.DriverManager.getConnection(url)) {
                ha.initBase(c);
            }
        }

        ha.initCluster();

        ha.loadData(ha.getConnection());

        Thread.sleep(1000);

        System.out.println("-------CLUSTER--------------------------------------");
        ha.checkData(ha.getConnection());
        System.out.println("----------------------------------------------------");
        

        ha.closeConnecton();

        for (String url : URLS) {
            try (java.sql.Connection c = java.sql.DriverManager.getConnection(url)) {
                ha.checkData(c);
            }
        }
        System.out.println("finish ... " + (System.currentTimeMillis() - t) + " ms");

        //Thread.sleep(1000 * 60 * 5);

        System.exit(0);

    }

    public void initBases() throws Exception {

        for (String url : URLS) {
            long t0 = System.currentTimeMillis();
            try (java.sql.Connection c = java.sql.DriverManager.getConnection(url)) {
                c.createStatement().execute("DROP TABLE IF EXISTS tb_test");
                c.createStatement().execute("CREATE TABLE IF NOT EXISTS tb_test ()");

                System.out.println("init " + c.getMetaData().getURL() + " " + (System.currentTimeMillis() - t0));

            } catch (java.sql.SQLException ex) {
                ex.printStackTrace();
                Thread.sleep(1000);
            }

        }
    }

    public void initCluster() throws InterruptedException {
        try {

            java.util.ArrayList<net.sf.hajdbc.sql.DriverDatabase> bases = new java.util.ArrayList<net.sf.hajdbc.sql.DriverDatabase>();

            int i = 0;
            for (String url : URLS) {
                net.sf.hajdbc.sql.DriverDatabase base = new net.sf.hajdbc.sql.DriverDatabase();
                base.setId("base" + i++);
                base.setLocation(url);

                bases.add(base);
            }

            net.sf.hajdbc.sql.DriverDatabaseClusterConfiguration config = new net.sf.hajdbc.sql.DriverDatabaseClusterConfiguration();

            config.setDatabases(bases);

            config.setDialectFactory(new net.sf.hajdbc.dialect.postgresql.PostgreSQLDialectFactory());

            config.setTransactionMode(net.sf.hajdbc.sql.TransactionModeEnum.SERIAL);

            config.setCurrentDateEvaluationEnabled(true);
            config.setCurrentTimeEvaluationEnabled(true);
            config.setCurrentTimestampEvaluationEnabled(true);
            config.setRandEvaluationEnabled(true);
            config.setSequenceDetectionEnabled(true);
            config.setIdentityColumnDetectionEnabled(true);

            config.setDatabaseMetaDataCacheFactory(new net.sf.hajdbc.cache.eager.SharedEagerDatabaseMetaDataCacheFactory());
            config.setBalancerFactory(new net.sf.hajdbc.balancer.simple.SimpleBalancerFactory());
            config.setStateManagerFactory(new net.sf.hajdbc.state.simple.SimpleStateManagerFactory());
            //config.setAutoActivationExpression(new net.sf.hajdbc.util.concurrent.cron.CronExpression("0 0/1 * 1/1 * ? *"));

            config.setDefaultSynchronizationStrategy("full");

            java.util.ArrayList<net.sf.hajdbc.SynchronizationStrategy> listStrategy = new java.util.ArrayList<net.sf.hajdbc.SynchronizationStrategy>() {
                {
                    add(new net.sf.hajdbc.sync.FullSynchronizationStrategy());
                    add(new net.sf.hajdbc.sync.DifferentialSynchronizationStrategy());
                    add(new net.sf.hajdbc.sync.FastDifferentialSynchronizationStrategy());
                    add(new net.sf.hajdbc.sync.PerTableSynchronizationStrategy(new net.sf.hajdbc.sync.FullSynchronizationStrategy()));
                    add(new net.sf.hajdbc.sync.PassiveSynchronizationStrategy());
                }
            };

            config.setSynchronizationStrategyMap(new java.util.TreeMap<>() {
                {
                    for (net.sf.hajdbc.SynchronizationStrategy e : listStrategy) {
                        put(e.getId(), e);
                    }
                }
            });

            net.sf.hajdbc.sql.Driver.setConfigurationFactory("cluster", new net.sf.hajdbc.SimpleDatabaseClusterConfigurationFactory<java.sql.Driver, net.sf.hajdbc.sql.DriverDatabase>(config));

            connection = java.sql.DriverManager.getConnection("jdbc:ha-jdbc:cluster", "sa", "");

        } catch (Exception ex) {
            ex.printStackTrace();
            Thread.sleep(1000);
        }
    }

    private java.sql.Connection connection;

    private java.sql.Connection getConnection() {
        return connection;
    }

    private void closeConnecton() throws InterruptedException {

        //DatabaseCluster.stop();
        try {
            if (connection != null) {
                connection.close();
            }
        } catch (Exception ex) {
            ex.printStackTrace();
            Thread.sleep(1000);
        }
    }

    public void checkData(java.sql.Connection connection) throws InterruptedException {
        try {
            System.out.println("check data: " + connection.getCatalog());

            try (java.sql.ResultSet rsTables = connection.createStatement().executeQuery(""
                    + "SELECT * "
                    + "  FROM information_schema.tables"
                    + " ORDER BY table_name")) {

                while (rsTables.next()) {
                    String tableName = rsTables.getString("table_name");
                    if (tableName.toLowerCase().startsWith("tb_test")) {
                        System.out.println("\t" + rsTables.getString("table_name"));

                        try (java.sql.ResultSet rs = connection.createStatement().executeQuery(""
                                + "SELECT  * "
                                + "  FROM " + tableName
                                + " ORDER BY 1")) {

                            java.sql.ResultSetMetaData metaData = rs.getMetaData();
                            for (int col = 0; col < metaData.getColumnCount(); col++) {
                                if (col > 0) {
                                    System.out.print(", ");
                                }
                                System.out.printf("%s (%s)", metaData.getColumnName(col + 1), metaData.getColumnTypeName(col + 1));
                            }
                            System.out.println();
                            while (rs.next()) {
                                for (int col = 0; col < metaData.getColumnCount(); col++) {
                                    if (col > 0) {
                                        System.out.print("\t");
                                    }
                                    System.out.printf("%s", rs.getString(metaData.getColumnName(col + 1)));
                                }
                                System.out.println();
                            }
                        }
                    }

                }
            }
            System.out.println("--------------------");
        } catch (java.sql.SQLException ex) {
            ex.printStackTrace();
            Thread.sleep(1000);
        }

    }

    private void initBase(java.sql.Connection connection) throws InterruptedException {
        try {
            System.out.println("Init base: " + connection.getCatalog());

            connection.createStatement().execute("DROP TABLE IF EXISTS tb_test");
            //connection.createStatement().execute("CREATE SCHEMA IF NOT EXISTS PUBLIC");
            connection.createStatement().execute("CREATE TABLE IF NOT EXISTS tb_test (id serial NOT NULL, CONSTRAINT pk_test PRIMARY KEY (id))");
            connection.createStatement().execute("ALTER TABLE tb_test ADD COLUMN IF NOT EXISTS verstamp timestamp DEFAULT NOW()");
            connection.createStatement().execute("ALTER TABLE tb_test ADD COLUMN IF NOT EXISTS data varchar");

        } catch (java.sql.SQLException ex) {
            ex.printStackTrace();
            Thread.sleep(1000);
        }

    }

    private void loadData(java.sql.Connection connection) throws Exception {
        try {
            connection.createStatement().execute(""
                    + "INSERT INTO tb_test (data) "
                    + "             VALUES (CAST(NOW() AS varchar))");

            connection.createStatement().execute(""
                    + "INSERT INTO tb_test (data) "
                    + "             VALUES (CAST(RANDOM() AS varchar))");

        } catch (java.sql.SQLException ex) {
            ex.printStackTrace();
            Thread.sleep(1000);
        }

    }

}

private Prepared parseShow() {
...
        } else if (readIf("SEARCH_PATH")) {
            buff.append("'");
            if (session.getSchemaSearchPath() == null || session.getSchemaSearchPath().length == 0) {
                buff.append(session.getCurrentSchemaName());
            } else {
                for (String sn : session.getSchemaSearchPath()) {
                    buff.append(sn);
                    buff.append(", ");
                }
                buff.setLength(buff.length() - 2);
            }
            buff.append("' AS SEARCH_PATH FROM DUAL");
...

package ha.jdbc_test;

public class HAJDBC_Test {

    private static final java.util.ArrayList<String> URLS = new java.util.ArrayList<String>() {
        {
            add("jdbc:postgresql://127.0.0.1/test_pg?user=postgres&password=secret");
            add("jdbc:h2:./test_h2;MODE=PostgreSQL;DATABASE_TO_LOWER=TRUE;USER=sa;PASSWORD=");
        }
    };

    public HAJDBC_Test() {
    }

    public static void main(String[] args) throws Exception {

        long t = System.currentTimeMillis();

        HAJDBC_Test ha = new HAJDBC_Test();

        for (String url : URLS) {
            try (java.sql.Connection c = java.sql.DriverManager.getConnection(url)) {
                ha.initBase(c);
            }
        }

        ha.initCluster();
        ha.loadData(ha.getConnection());

        System.out.println("=======CLUSTER=================================================");
        ha.checkData(ha.getConnection());
        ha.testProvodki(ha.getConnection());
        ha.closeConnecton();
        System.out.println("===============================================================");

        for (String url : URLS) {
            try (java.sql.Connection c = java.sql.DriverManager.getConnection(url)) {
                ha.checkData(c);
            }
        }

        System.out.println("old");
        System.out.println("old");
        System.out.println("old");
        System.out.println("new");

        System.out.println("=======SINGLE DATABASE=========================================");
        for (String url : URLS) {
            try (java.sql.Connection c = java.sql.DriverManager.getConnection(url)) {
                ha.testProvodki(c);
            }
        }

        System.out.println("finish ... " + (System.currentTimeMillis() - t) + " ms");

        //Thread.sleep(1000 * 60 * 5);
        System.exit(0);

    }

    public void initBases() throws Exception {

        for (String url : URLS) {
            long t0 = System.currentTimeMillis();
            try (java.sql.Connection c = java.sql.DriverManager.getConnection(url)) {
                c.createStatement().execute("DROP TABLE IF EXISTS tb_test");
                c.createStatement().execute("CREATE TABLE IF NOT EXISTS tb_test ()");

                System.out.println("init " + c.getMetaData().getURL() + " " + (System.currentTimeMillis() - t0));

            } catch (java.sql.SQLException ex) {
                ex.printStackTrace();
                Thread.sleep(1000);
            }

        }
    }

    public void initCluster() throws InterruptedException {
        try {

            java.util.ArrayList<net.sf.hajdbc.sql.DriverDatabase> bases = new java.util.ArrayList<net.sf.hajdbc.sql.DriverDatabase>();

            int i = 0;
            for (String url : URLS) {
                net.sf.hajdbc.sql.DriverDatabase base = new net.sf.hajdbc.sql.DriverDatabase();
                base.setId("base" + i++);
                base.setLocation(url);

                bases.add(base);
            }

            net.sf.hajdbc.sql.DriverDatabaseClusterConfiguration config = new net.sf.hajdbc.sql.DriverDatabaseClusterConfiguration();

            config.setDatabases(bases);

            config.setDialectFactory(new net.sf.hajdbc.dialect.postgresql.PostgreSQLDialectFactory());

            config.setTransactionMode(net.sf.hajdbc.sql.TransactionModeEnum.SERIAL);

            config.setCurrentDateEvaluationEnabled(true);
            config.setCurrentTimeEvaluationEnabled(true);
            config.setCurrentTimestampEvaluationEnabled(true);
            config.setRandEvaluationEnabled(true);
            config.setSequenceDetectionEnabled(true);
            config.setIdentityColumnDetectionEnabled(true);

            config.setDatabaseMetaDataCacheFactory(new net.sf.hajdbc.cache.eager.SharedEagerDatabaseMetaDataCacheFactory());
            config.setBalancerFactory(new net.sf.hajdbc.balancer.simple.SimpleBalancerFactory());
            config.setStateManagerFactory(new net.sf.hajdbc.state.simple.SimpleStateManagerFactory());
            //config.setAutoActivationExpression(new net.sf.hajdbc.util.concurrent.cron.CronExpression("0 0/1 * 1/1 * ? *"));

            config.setDefaultSynchronizationStrategy("full");

            java.util.ArrayList<net.sf.hajdbc.SynchronizationStrategy> listStrategy = new java.util.ArrayList<net.sf.hajdbc.SynchronizationStrategy>() {
                {
                    add(new net.sf.hajdbc.sync.FullSynchronizationStrategy());
                    add(new net.sf.hajdbc.sync.DifferentialSynchronizationStrategy());
                    add(new net.sf.hajdbc.sync.FastDifferentialSynchronizationStrategy());
                    add(new net.sf.hajdbc.sync.PerTableSynchronizationStrategy(new net.sf.hajdbc.sync.FullSynchronizationStrategy()));
                    add(new net.sf.hajdbc.sync.PassiveSynchronizationStrategy());
                }
            };

            config.setSynchronizationStrategyMap(new java.util.TreeMap<>() {
                {
                    for (net.sf.hajdbc.SynchronizationStrategy e : listStrategy) {
                        put(e.getId(), e);
                    }
                }
            });

            net.sf.hajdbc.sql.Driver.setConfigurationFactory("cluster", new net.sf.hajdbc.SimpleDatabaseClusterConfigurationFactory<java.sql.Driver, net.sf.hajdbc.sql.DriverDatabase>(config));

            connection = java.sql.DriverManager.getConnection("jdbc:ha-jdbc:cluster", "sa", "");

        } catch (Exception ex) {
            ex.printStackTrace();
            Thread.sleep(1000);
        }
    }

    private java.sql.Connection connection;

    private java.sql.Connection getConnection() {
        return connection;
    }

    private void closeConnecton() throws InterruptedException {

        //DatabaseCluster.stop();
        try {
            if (connection != null) {
                connection.close();
            }
        } catch (Exception ex) {
            ex.printStackTrace();
            Thread.sleep(1000);
        }
    }

    public void checkData(java.sql.Connection connection) throws InterruptedException {
        try {
            System.out.println("check data: " + connection.getCatalog());

            try (java.sql.ResultSet rsTables = connection.createStatement().executeQuery(""
                    + "SELECT * "
                    + "  FROM information_schema.tables"
                    + " ORDER BY table_name")) {

                while (rsTables.next()) {
                    String tableName = rsTables.getString("table_name");
                    if (tableName.toLowerCase().startsWith("tb_test")) {
                        System.out.println("\t" + rsTables.getString("table_name"));

                        try (java.sql.ResultSet rs = connection.createStatement().executeQuery(""
                                + "SELECT  * "
                                + "  FROM " + tableName
                                + " ORDER BY 1")) {

                            java.sql.ResultSetMetaData metaData = rs.getMetaData();
                            for (int col = 0; col < metaData.getColumnCount(); col++) {
                                if (col > 0) {
                                    System.out.print(", ");
                                }
                                System.out.printf("%s (%s)", metaData.getColumnName(col + 1), metaData.getColumnTypeName(col + 1));
                            }
                            System.out.println();
                            while (rs.next()) {
                                for (int col = 0; col < metaData.getColumnCount(); col++) {
                                    if (col > 0) {
                                        System.out.print("\t");
                                    }
                                    System.out.printf("%s", rs.getString(metaData.getColumnName(col + 1)));
                                }
                                System.out.println();
                            }
                        }
                    }

                }
            }
            System.out.println("--------------------");
        } catch (java.sql.SQLException ex) {
            ex.printStackTrace();
            Thread.sleep(1000);
        }

    }

    public void initBase(java.sql.Connection connection) throws InterruptedException {
        try {
            System.out.println("Init base: " + connection.getCatalog());

            connection.createStatement().execute("DROP TABLE IF EXISTS tb_test");
            //connection.createStatement().execute("CREATE SCHEMA IF NOT EXISTS PUBLIC");
            connection.createStatement().execute("CREATE TABLE IF NOT EXISTS tb_test (id serial NOT NULL, CONSTRAINT pk_test PRIMARY KEY (id))");
            connection.createStatement().execute("ALTER TABLE tb_test ADD COLUMN IF NOT EXISTS verstamp timestamp DEFAULT NOW()");
            connection.createStatement().execute("ALTER TABLE tb_test ADD COLUMN IF NOT EXISTS data varchar");

        } catch (java.sql.SQLException ex) {
            ex.printStackTrace();
            Thread.sleep(1000);
        }

    }

    public void loadData(java.sql.Connection connection) throws Exception {
        try {
            connection.createStatement().execute(""
                    + "INSERT INTO tb_test (data) "
                    + "             VALUES (CAST(NOW() AS varchar))");

            connection.createStatement().execute(""
                    + "INSERT INTO tb_test (data) "
                    + "             VALUES (CAST(RANDOM() AS varchar))");

        } catch (java.sql.SQLException ex) {
            ex.printStackTrace();
            Thread.sleep(1000);
        }

    }

    public void testProvodki(java.sql.Connection connection) throws Exception {

        long t_startTest = System.currentTimeMillis();

        System.out.println("===============================================================");
        System.out.println("!!!!!!!!!!!!!!!!!testProvodki!!!!!!!!!!!!!!!!!!!!!!!!!!!!!");
        System.out.println("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~");

        System.out.println(" start testProvodki: " + connection.getCatalog() + " " + t_startTest);

        /*, tmp_temp*/ /*need fix source h2 multiple drop*/
        connection.createStatement().execute(""
                + "DROP TABLE IF EXISTS tmp_provodki");
        connection.createStatement().execute(""
                + "DROP TABLE IF EXISTS tmp_temp");

        /*unlogged*/ /*need fix source unlogged/temp table*/
        connection.createStatement().execute(""
                + "CREATE TABLE tmp_provodki \n"
                + "   (from_id  smallint,\n"
                + "    to_id    smallint,\n"
                + "    amount   bigint);"
        );

        long t_startInsert = System.currentTimeMillis();

        System.out.println("     start insert tmp_provodki: " + connection.getCatalog() + " " + (t_startInsert));

        //generate_series(1, 1000000, 1) //--need fix hajdbc --#%$#@Q!#~!@#~ map foreach bath memory
        //connection.createStatement().execute("INSERT INTO tmp_provodki (SELECT random()*89  + 1, random()*89 + 1, random()*100 FROM generate_series(1, 1000000, 1))");
        //"nj rfgtw"
        java.sql.PreparedStatement stmt = connection.prepareStatement("INSERT INTO tmp_provodki VALUES (?, ?, ?)");

        for (int i = 0; i < 10_000_000; i++) {
            stmt.setInt(1, (int) (Math.random() * 89));
            stmt.setInt(2, (int) (Math.random() * 89));
            stmt.setInt(3, (int) (Math.random() * 100));
            stmt.addBatch();
            if (i % 10_000 == 0) {
                stmt.executeBatch();
            }
        }

        //java.sql.Statement stmt = connection.createStatement();
        //for (int i = 0; i < 10; i++) {
        //    stmt.addBatch(String.format("INSERT INTO tmp_provodki VALUES (%d, %d, %d)", (int) (Math.random() * 89), (int) (Math.random() * 89), (int) (Math.random() * 100)));
        //}
        stmt.executeBatch();
        System.out.println("     finish insert tmp_provodki: " + connection.getCatalog() + " " + (System.currentTimeMillis() - t_startInsert) + "");

        System.out.println("     __________________________________________________________");

        long t_startGroup = System.currentTimeMillis();
        System.out.println("         start group tmp_provodki: " + connection.getCatalog() + " >>>" + System.currentTimeMillis());
        connection.createStatement().execute(""
                + ""
                + "CREATE TEMP TABLE results AS ( \n"
                + "SELECT q.from_id, pos - neg    \n"
                + "   FROM (SELECT from_id, sum(amount) as neg           \n"
                + "          FROM tmp_provodki          \n"
                + "         GROUP BY from_id) AS q   \n"
                + "   LEFT JOIN (SELECT to_id, sum(amount) as pos\n"
                + "                FROM tmp_provodki\n"
                + "               GROUP BY to_id) AS w ON q.from_id = w.to_id\n"
                + " )"
        );
        System.out.println("         finish group tmp_provodki: " + connection.getCatalog() + " " + (System.currentTimeMillis() - t_startInsert) + "");
        System.out.println("     __________________________________________________________");
        System.out.println("finish testProvodki: " + connection.getCatalog() + " " + (System.currentTimeMillis() - t_startTest) + "");
        System.out.println("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~");
        System.out.println("!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!");
        System.out.println("===============================================================");
    }

}