presto-base-jdbc/src/main/java/io/prestosql/plugin/jdbc/JdbcPageSink.java/JdbcPageSink/JdbcPageSink
class JdbcPageSink:    public JdbcPageSink(ConnectorSession session, JdbcOutputTableHandle handle, JdbcClient jdbcClient)
    {
        try {
            connection = jdbcClient.getConnection(JdbcIdentity.from(session), handle);
        }
        catch (SQLException e) {
            throw new PrestoException(JDBC_ERROR, e);
        }

        try {
            connection.setAutoCommit(false);
            statement = connection.prepareStatement(jdbcClient.buildInsertSql(handle));
        }
        catch (SQLException e) {
            closeWithSuppression(connection, e);
            throw new PrestoException(JDBC_ERROR, e);
        }

        columnTypes = handle.getColumnTypes();

        if (!handle.getJdbcColumnTypes().isPresent()) {
            List<WriteMapping> writeMappings = columnTypes.stream()
                    .map(type -> {
                        WriteMapping writeMapping = jdbcClient.toWriteMapping(session, type);
                        WriteFunction writeFunction = writeMapping.getWriteFunction();
                        verify(
                                type.getJavaType() == writeFunction.getJavaType(),
                                "Presto type %s is not compatible with write function %s accepting %s",
                                type,
                                writeFunction,
                                writeFunction.getJavaType());
                        return writeMapping;
                    })
                    .collect(toImmutableList());

            columnWriters = writeMappings.stream()
                    .map(WriteMapping::getWriteFunction)
                    .collect(toImmutableList());

            nullWriters = writeMappings.stream()
                    .map(WriteMapping::getWriteNullFunction)
                    .collect(toImmutableList());
        }
        else {
            List<ColumnMapping> columnMappings = handle.getJdbcColumnTypes().get().stream()
                    .map(typeHandle -> {
                        Optional<ColumnMapping> columnMapping = jdbcClient.toPrestoType(session, connection, typeHandle);
                        checkState(columnMapping.isPresent(), "missing column mapping");
                        return columnMapping.get();
                    })
                    .collect(toImmutableList());

            columnWriters = columnMappings.stream()
                    .map(ColumnMapping::getWriteFunction)
                    .collect(toImmutableList());

            nullWriters = columnMappings.stream()
                    .map(ColumnMapping::getWriteNullFunction)
                    .collect(toImmutableList());
        }
    }