Contents

package org.embulk.input.postgresql;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import org.slf4j.Logger;
import org.embulk.spi.Exec;
import org.embulk.input.jdbc.JdbcInputConnection;

public class PostgreSQLInputConnection
        extends JdbcInputConnection
{
    private final Logger logger = Exec.getLogger(PostgreSQLInputConnection.class);

    public PostgreSQLInputConnection(Connection connection, String schemaName)
            throws SQLException
    {
        super(connection, schemaName);
    }

    @Override
    protected CursorSelect newBatchSelect(String select, int fetchRows, int queryTimeout) throws SQLException
    {
        executeUpdate("DECLARE cur NO SCROLL CURSOR FOR "+select);

        String fetchSql = "FETCH FORWARD "+fetchRows+" FROM cur";
        // Because socketTimeout is set in Connection, don't need to set quertyTimeout.
        return new CursorSelect(fetchSql, connection.prepareStatement(fetchSql));
    }

    public class CursorSelect
            implements BatchSelect
    {
        private final String fetchSql;
        private final PreparedStatement fetchStatement;

        public CursorSelect(String fetchSql, PreparedStatement fetchStatement) throws SQLException
        {
            this.fetchSql = fetchSql;
            this.fetchStatement = fetchStatement;
        }

        public ResultSet fetch() throws SQLException
        {
            logger.info("SQL: " + fetchSql);
            long startTime = System.currentTimeMillis();

            ResultSet rs = fetchStatement.executeQuery();

            double seconds = (System.currentTimeMillis() - startTime) / 1000.0;
            logger.info(String.format("> %.2f seconds", seconds));
            return rs;
        }

        public void close() throws SQLException
        {
            // TODO close?
        }
    }
}

Version data entries

7 entries across 7 versions & 1 rubygems

Version	Path
embulk-input-postgresql-0.7.2	src/main/java/org/embulk/input/postgresql/PostgreSQLInputConnection.java
embulk-input-postgresql-0.7.1	src/main/java/org/embulk/input/postgresql/PostgreSQLInputConnection.java
embulk-input-postgresql-0.7.0	src/main/java/org/embulk/input/postgresql/PostgreSQLInputConnection.java
embulk-input-postgresql-0.6.4	src/main/java/org/embulk/input/postgresql/PostgreSQLInputConnection.java
embulk-input-postgresql-0.6.3	src/main/java/org/embulk/input/postgresql/PostgreSQLInputConnection.java
embulk-input-postgresql-0.6.2	src/main/java/org/embulk/input/postgresql/PostgreSQLInputConnection.java
embulk-input-postgresql-0.6.1	src/main/java/org/embulk/input/postgresql/PostgreSQLInputConnection.java