D-Net

package eu.dnetlib.data.mapreduce.hbase.statsExport.daos;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FSDataInputStream;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

import org.apache.log4j.Logger;

import org.junit.Test;

import org.postgresql.copy.CopyManager;

import org.postgresql.core.BaseConnection;

import org.springframework.jdbc.core.JdbcTemplate;

import org.springframework.jdbc.core.simple.SimpleJdbcCall;

import org.springframework.jdbc.datasource.DriverManagerDataSource;

import java.io.FileInputStream;

import java.io.InputStream;

import java.sql.CallableStatement;

import java.sql.Connection;

import java.sql.ResultSet;

import java.sql.ResultSetMetaData;

/**

 * @author eri

 *         <p/>

 *         DB DAO with methods for creating the temporary shadow schema of the

 *         statsDB, executing extra inserts, creating views and indexes .

 */

public class StatsDAO {

    private org.springframework.jdbc.datasource.DriverManagerDataSource statsDatasource;

    private Logger log = Logger.getLogger(this.getClass());

    private long startTime;

    private long endtime;

    public StatsDAO(String dbUrl, String dbUser, String dbPassword, String dbDriver) {

        statsDatasource = new DriverManagerDataSource(dbUrl, dbUser, dbPassword);

        statsDatasource.setDriverClassName(dbDriver);

    }

    /**

     * Reads create schema script from a file and executes it. Used to create a

     * temporary Shadow schema for data import

     *

     * @throws Exception

     */

    public void createSchema() throws Exception {

        InputStream in = ClassLoader.getSystemResourceAsStream("eu/dnetlib/data/mapreduce/hbase/statsExport/stats_db_schema.sql");

        byte[] b = new byte[in.available()];

        in.read(b);

        String q = new String(b);

        in.close();

        Connection con = null;

        log.info(" Creating Shadow Schema Database   ....");

        try {

            con = statsDatasource.getConnection();

            CallableStatement st = con.prepareCall(q);

            st.execute();

            st.close();

        } catch (Exception e) {

            log.error("Could not   Create Tables in  Temporary database  :" + e.getMessage());

            throw new Exception("Could not Create Tables in Temporary database  :", e);

        } finally {

            con.close();

        }

        log.info("Shadow Schema created.");

    }

    /**

     * Sets the default schema (search path ) for the sqoop user.

     *

     * @param user

     * @throws Exception

     */

    public void setSearchPathDB(String user) throws Exception {

        String q = "ALTER ROLE " + user + " SET search_path TO shadow,public; ";

        Connection con = null;

        try {

            con = statsDatasource.getConnection();

            CallableStatement st = con.prepareCall(q);

            st.execute();

            st.close();

        } catch (Exception e) {

            log.error("Could not change default schema for user :" + e.getMessage());

            throw new Exception("Could not change default schema for user ", e);

        } finally {

            con.close();

        }

        log.info(" Default user schema set to shadow");

    }

    /**

     * Keeps a backup of the current public schema and then replaces it with the new  Shadow one.

     *

     * @throws Exception

     */

    public void renameSchema() throws Exception {

        Connection con = statsDatasource.getConnection();

        String q = "DROP SCHEMA IF EXISTS backup CASCADE;" +

                "DROP SCHEMA IF EXISTS public CASCADE;" +

                " ALTER SCHEMA shadow RENAME TO public; ";

        try {

            CallableStatement st = con.prepareCall(q);

            log.info("Renaming  Shadow  schema to  Public,,,");

            st.execute();

            st.close();

            con.close();

            log.info("Renaming done");

        } catch (Exception e) {

            log.error("Could not rename schema " + e.getMessage());

            throw new Exception("Could not rename schema  ", e);

        } finally {

            con.close();

        }

        log.info("Schema renamed");

    }

    public void addArrayColumns() throws Exception {

        log.info("Adding new columns ...");

        Connection con = statsDatasource.getConnection();

        startTime = System.currentTimeMillis();

        String q = "{call shadow.create_arrays()}";

        CallableStatement st = con.prepareCall(q);

        st.execute();

        st.close();

        con.close();

        endtime = System.currentTimeMillis();

        log.info("Time to add columns: " + ((endtime - startTime) / 60000) + " minutes ");

    }

    public void cleanTables() throws Exception {

        log.info(" Cleaning Tables...");

        Connection con = statsDatasource.getConnection();

        startTime = System.currentTimeMillis();

        String q = "CREATE TABLE \"shadow\".rd_distinct AS SELECT DISTINCT * FROM \"shadow\".result_datasources;";

        CallableStatement st = con.prepareCall(q);

        st.execute();

        st.close();

        q = "TRUNCATE \"shadow\".result_datasources;";

        st = con.prepareCall(q);

        st.execute();

        st.close();

        q = "INSERT INTO \"shadow\".result_datasources SELECT * FROM \"shadow\".rd_distinct;";

        st = con.prepareCall(q);

        st.execute();

        st.close();

        q = "DROP TABLE \"shadow\".rd_distinct;";

        st = con.prepareCall(q);

        st.execute();

        st.close();

        con.close();

        endtime = System.currentTimeMillis();

        log.info("Time to clean tables : " + ((endtime - startTime) / 60000) + " minutes ");

    }

    /**

     * Create charts.

     *

     * @throws Exception

     */

    public void createCharts() throws Exception {

        log.info(" Creating Chart Tables...");

        Connection con = statsDatasource.getConnection();

        startTime = System.currentTimeMillis();

        String q = "{call shadow.create_charts()}";

        CallableStatement st = con.prepareCall(q);

        st.execute();

        st.close();

        con.close();

        endtime = System.currentTimeMillis();

        log.info("Time to create chart tables: " + ((endtime - startTime) / 60000) + " minutes ");

    }

    /**

     * Create chart indexes.

     *

     * @throws Exception

     */

    public void createChartIndexes() throws Exception {

        log.info(" Create Chart Indexes...");

        Connection con = statsDatasource.getConnection();

        startTime = System.currentTimeMillis();

        String q = "{call shadow.create_chart_indexes()}";

        CallableStatement st = con.prepareCall(q);

        st.execute();

        st.close();

        con.close();

        endtime = System.currentTimeMillis();

        log.info("Time to create chart indexes : " + ((endtime - startTime) / 60000) + " minutes ");

    }

    /**

     * Builds indexes.

     *

     * @throws Exception

     */

    public void buildIndexes() throws Exception {

        log.info(" Building Database Indexes...");

        Connection con = statsDatasource.getConnection();

        startTime = System.currentTimeMillis();

        String q = "{call shadow.create_indexes()}";

        CallableStatement st = con.prepareCall(q);

        st.execute();

        st.close();

        con.close();

        endtime = System.currentTimeMillis();

        log.info("Time to build indexes : " + ((endtime - startTime) / 60000) + " minutes ");

    }

    /**

     * Builds views.

     *

     * @throws Exception

     */

    public void buildViews() throws Exception {

        log.info(" Building Database Views...");

        Connection con = statsDatasource.getConnection();

        startTime = System.currentTimeMillis();

        String q = "{call shadow.create_views()}";

        CallableStatement st = con.prepareCall(q);

        st.execute();

        st.close();

        con.close();

        endtime = System.currentTimeMillis();

        log.info("Time to build Views : " + ((endtime - startTime) / 60000) + " minutes ");

    }

    /**

     * Executes extra inserts ( table updates )

     *

     * @throws Exception

     */

    public void executeExtraInserts() throws Exception {

        log.info("  Executing Extra Inserts...");

        startTime = System.currentTimeMillis();

        populateDefaults();

        //update_project_results();

        update_project_has_pubs();

        update_project_pubs_count();

        update_project_delated_pubs();

        update_project_daysforlastpub();

        update_project_delayed_pubs();

        cleanUp();

        endtime = System.currentTimeMillis();

        log.info("Total time for Extra Inserts : " + ((endtime - startTime) / 60000) + " minutes ");

    }

    public void populateDefaults() throws Exception {

        startTime = System.currentTimeMillis();

        log.info("  Populating defaults table and Updating Datasources...");

        Connection con = statsDatasource.getConnection();

        startTime = System.currentTimeMillis();

        String q = "{call shadow.extra_defaults_datasource()}";

        CallableStatement st = con.prepareCall(q);

        st.execute();

        st.close();

        con.close();

        //JdbcTemplate jdbcTemplate = new JdbcTemplate(statsDatasource);

        //jdbcTemplate.setResultsMapCaseInsensitive(true);

        //new SimpleJdbcCall(statsDatasource).withProcedureName("extra_defaults_datasource");

        endtime = System.currentTimeMillis();

        log.info("Time  for   Populating defaults : " + ((endtime - startTime) / 60000) + " minutes ");

    }

    public void update_project_results() throws Exception {

        startTime = System.currentTimeMillis();

        log.info("   Updating Project Results...");

        Connection con = statsDatasource.getConnection();

        startTime = System.currentTimeMillis();

        String q = "{call shadow.update_project_results()}";

        CallableStatement st = con.prepareCall(q);

        st.execute();

        st.close();

        con.close();

        //JdbcTemplate jdbcTemplate = new JdbcTemplate(statsDatasource);

        //jdbcTemplate.setResultsMapCaseInsensitive(true);

        //new SimpleJdbcCall(statsDatasource).withProcedureName("update_project_results");

        log.info("Done.");

        endtime = System.currentTimeMillis();

        log.info("Time  for  Updating Project Results: " + ((endtime - startTime) / 60000) + " minutes ");

    }

    public void update_project_has_pubs() throws Exception {

        startTime = System.currentTimeMillis();

        log.info("   Updating Project with Publications..");

        Connection con = statsDatasource.getConnection();

        startTime = System.currentTimeMillis();

        String q = "{call shadow.project_has_pubs()}";

        CallableStatement st = con.prepareCall(q);

        st.execute();

        st.close();

        con.close();

        //JdbcTemplate jdbcTemplate = new JdbcTemplate(statsDatasource);

        //jdbcTemplate.setResultsMapCaseInsensitive(true);

        //new SimpleJdbcCall(statsDatasource).withProcedureName("project_has_pubs");

        log.info("Done.");

        endtime = System.currentTimeMillis();

        log.info("Time  for Updating Project with Publications : " + ((endtime - startTime) / 60000) + " minutes ");

    }

    public void update_project_pubs_count() throws Exception {

        startTime = System.currentTimeMillis();

        log.info("   Updating Project  Publications Count..");

        Connection con = statsDatasource.getConnection();

        startTime = System.currentTimeMillis();

        String q = "{call shadow.project_pubs_count()}";

        CallableStatement st = con.prepareCall(q);

        st.execute();

        st.close();

        con.close();

        //JdbcTemplate jdbcTemplate = new JdbcTemplate(statsDatasource);

        //jdbcTemplate.setResultsMapCaseInsensitive(true);

        //new SimpleJdbcCall(statsDatasource).withProcedureName("project_pubs_count");

        log.info("Done.");

        endtime = System.currentTimeMillis();

        log.info("Time  for Updating Project  Publications Count : " + ((endtime - startTime) / 60000) + " minutes ");

    }

    public void update_project_delated_pubs() throws Exception {

        startTime = System.currentTimeMillis();

        log.info("   Updating Project  Delayed Publications...");

        Connection con = statsDatasource.getConnection();

        startTime = System.currentTimeMillis();

        String q = "{call shadow.project_delayedpubs()}";

        CallableStatement st = con.prepareCall(q);

        st.execute();

        st.close();

        con.close();

        //JdbcTemplate jdbcTemplate = new JdbcTemplate(statsDatasource);

        //jdbcTemplate.setResultsMapCaseInsensitive(true);

        //new SimpleJdbcCall(statsDatasource).withProcedureName("project_delayedpubs");

        log.info("Done.");

        endtime = System.currentTimeMillis();

        log.info("Time  for Updating Project  Delayed Publications : " + ((endtime - startTime) / 60000) + " minutes ");

    }

    public void update_project_daysforlastpub() throws Exception {

        startTime = System.currentTimeMillis();

        log.info("Updating Project Days For Last Publication...");

        Connection con = statsDatasource.getConnection();

        startTime = System.currentTimeMillis();

        String q = "{call shadow.project_daysforlastpub()}";

        CallableStatement st = con.prepareCall(q);

        st.execute();

        st.close();

        con.close();

        //JdbcTemplate jdbcTemplate = new JdbcTemplate(statsDatasource);

        //jdbcTemplate.setResultsMapCaseInsensitive(true);

        //new SimpleJdbcCall(statsDatasource).withProcedureName("project_daysforlastpub");

        log.info("Done.");

        endtime = System.currentTimeMillis();

        log.info("Time  for Updating Project Days For Last Publication   : " + ((endtime - startTime) / 60000) + " minutes ");

    }

    public void update_project_delayed_pubs() throws Exception {

        startTime = System.currentTimeMillis();

        log.info("   Updating Project  Delayed  Publications...");

        Connection con = statsDatasource.getConnection();

        startTime = System.currentTimeMillis();

        String q = "{call shadow.project_delayed()}";

        CallableStatement st = con.prepareCall(q);

        st.execute();

        st.close();

        con.close();

        //JdbcTemplate jdbcTemplate = new JdbcTemplate(statsDatasource);

        //jdbcTemplate.setResultsMapCaseInsensitive(true);

        //new SimpleJdbcCall(statsDatasource).withProcedureName("project_delayed");

        log.info("Done.");

        endtime = System.currentTimeMillis();

        log.info("Time  for Updating  Project  Delayed  Publications : " + ((endtime - startTime) / 60000) + " minutes ");

    }

    public void insertCountries() throws Exception {

        startTime = System.currentTimeMillis();

        log.info("Inserting countries...");

        Connection con = statsDatasource.getConnection();

        startTime = System.currentTimeMillis();

        String q = "{call shadow.insert_countries()}";

        CallableStatement st = con.prepareCall(q);

        st.execute();

        st.close();

        con.close();

        log.info("Done.");

        endtime = System.currentTimeMillis();

        log.info("Time  for inserting countries: " + ((endtime - startTime) / 60000) + " minutes ");

    }

    public void backwardsCompatibility() throws Exception {

        startTime = System.currentTimeMillis();

        log.info("Ensuring backwards compatibility");

        Connection con = statsDatasource.getConnection();

        startTime = System.currentTimeMillis();

        String q = "{call shadow.backwards_compatibility()}";

        CallableStatement st = con.prepareCall(q);

        st.execute();

        st.close();

        con.close();

        log.info("Done.");

        endtime = System.currentTimeMillis();

        log.info("Time  for ensuring backwards compatibility: " + ((endtime - startTime) / 60000) + " minutes ");

    }

    public void cleanUp() throws Exception {

        startTime = System.currentTimeMillis();

        log.info("  Cleaning Up...");

        Connection con = statsDatasource.getConnection();

        startTime = System.currentTimeMillis();

        String q = "{call shadow.cleanTemps()}";

        CallableStatement st = con.prepareCall(q);

        st.execute();

        st.close();

        con.close();

        //JdbcTemplate jdbcTemplate = new JdbcTemplate(statsDatasource);

        //jdbcTemplate.setResultsMapCaseInsensitive(true);

        //new SimpleJdbcCall(statsDatasource).withProcedureName("cleanTemps");

        log.info("Done.");

        endtime = System.currentTimeMillis();

        log.info("Time  for  cleaning Up : " + ((endtime - startTime) / 60000) + " minutes ");

    }

    /**

     * Loads bulk data for Context( and category) tables as generated from the

     * IS xml

     *

     * @throws Exception

     */

    public void insertContexts(String fileDir) throws Exception {

        try {

            if (!fileDir.endsWith("/")) {

                fileDir += "/";

            }

            copy("context", fileDir + "context");

            copy("concept", fileDir + "concept");

            copy("category", fileDir + "category");

        } catch (Exception e) {

            log.error("Could not import  context in  new database  :" + e);

            throw new Exception("Could not import context in new database  :", e);

        }

        log.info(" Done.");

    }

    /**

     * Calls the COPY command for the given @table and inserts data from a file

     *

     * @param table

     * @throws Exception

     */

    public void copy(String table, String file) throws Exception {

        CopyManager copyManager;

        FileInputStream inputStream = null;

        FileSystem fs = FileSystem.get(new Configuration());

        Connection con = null;

        try {

            FSDataInputStream fin = fs.open(new Path(file));

            con = statsDatasource.getConnection();

            copyManager = new CopyManager((BaseConnection) con);

            inputStream = new FileInputStream(file);

            copyManager.copyIn("COPY " + table + " FROM STDIN ", fin);

            log.debug("Copy done for " + table);

            fin.close();

        } catch (Exception e) {

            log.error("Fail to execute copy command for table " + table + e);

            throw new Exception("Fail to execute copy command for table " + table, e);

        } finally {

            con.close();

            inputStream.close();

        }

    }

    public String getResults(ResultSet rs) throws Exception {

        String data = new String();

        try {

            log.info("Generated Report : ");

            ResultSetMetaData rsmd = rs.getMetaData();

            while (rs.next()) {

                for (int i = 1; i < rsmd.getColumnCount() - 1; i++) {

                    data += rsmd.getColumnName(i) + "," + rs.getInt(i) + "\n";

                }

            }

            log.info(" Generated Data :" + data);

            return data;

        } catch (Exception e) {

            log.error("Could not import  context in  new database :" + e);

            throw new Exception("Could not import context in new database :", e);

        } finally {

            rs.close();

        }

    }

    public DriverManagerDataSource getStatsDatasource() {

        return statsDatasource;

    }

    public void setStatsDatasource(DriverManagerDataSource statsDatasource) {

        this.statsDatasource = statsDatasource;

    }

}