Hologres:COPY

Comportamento do ON_CONFLICT por versão

• Antes da V3.0.4: O parâmetro ON_CONFLICT só tem efeito quando STREAM_MODE TRUE.

• V3.0.4 e posteriores: O ON_CONFLICT também entra em vigor quando STREAM_MODE FALSE, desde que o parâmetro GUC hg_experimental_copy_enable_on_conflict esteja ativado. Com STREAM_MODE FALSE, a opção UPDATE exige a gravação de todas as colunas.

• V3.1.1 e posteriores: Quando STREAM_MODE FALSE, o UPDATE suporta importações de colunas parciais (o parâmetro hg_experimental_copy_enable_on_conflict vem ativado por padrão).

SELECT
    query_id,
    query,
    extended_info
FROM
    hologres.hg_query_log
WHERE
    extended_info ->> 'source_trx' = '<transaction_id>'  -- Get the transaction ID from the trans_id field in the COPY log record
ORDER BY
    query_start;

-- Create the target table
CREATE TABLE copy_test (
    id    int,
    age   int,
    name  text
);

-- Import data from stdin
COPY copy_test FROM STDIN WITH DELIMITER AS ',' NULL AS '';
53444,24,wangming
55444,38,ligang
55444,38,luyong
\.

-- Verify
SELECT * FROM copy_test;

-- Create the target table
CREATE TABLE partsupp (
    ps_partkey     integer NOT NULL,
    ps_suppkey     integer NOT NULL,
    ps_availqty    integer NOT NULL,
    ps_supplycost  float   NOT NULL,
    ps_comment     text    NOT NULL
);

-- Import CSV from stdin
COPY partsupp FROM STDIN WITH DELIMITER '|' CSV;
1|2|3325|771.64|final theodolites
1|25002|8076|993.49|ven ideas
\.

-- Verify
SELECT * FROM partsupp;

psql -U <username> -p <port> -h <endpoint> -d <databasename> \
    -c "COPY <table> FROM STDIN WITH DELIMITER '|' CSV;" <<filename>;

01,01,name1
02,01,name2
03,01,name3
04,01,name4

package com.aliyun.hologram.test.jdbc;

import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.sql.*;
import java.util.Properties;

import org.postgresql.copy.CopyManager;
import org.postgresql.core.BaseConnection;

public class jdbcCopyFile {

    public static void main(String args[]) throws Exception {
        System.out.println(copyFromFile(getConnection(), "/Users/feng/Workspace/region.tbl", "region"));
    }

    public static Connection getConnection() throws Exception {
        Class.forName("org.postgresql.Driver");
        String url = "jdbc:postgresql://endpoint:port/dbname";
        Properties props = new Properties();
        // Store credentials in environment variables to avoid hardcoding them
        props.setProperty("user", "AAA");     // AccessKey ID
        props.setProperty("password", "BBB"); // AccessKey secret
        return DriverManager.getConnection(url, props);
    }

    /**
     * Streams a local file into Hologres via COPY FROM STDIN.
     *
     * @param connection  Active JDBC connection
     * @param filePath    Path to the local file
     * @param tableName   Target Hologres table
     * @return Number of rows imported
     */
    public static long copyFromFile(Connection connection, String filePath, String tableName)
            throws SQLException, IOException {
        long count = 0;
        FileInputStream fileInputStream = null;

        try {
            CopyManager copyManager = new CopyManager((BaseConnection) connection);
            fileInputStream = new FileInputStream(filePath);
            count = copyManager.copyIn("COPY " + tableName + " FROM STDIN delimiter '|' csv", fileInputStream);
        } finally {
            if (fileInputStream != null) {
                try {
                    fileInputStream.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
        return count;
    }
}

CREATE TABLE t0 (id int NOT NULL, name text, age int, primary key(id));

COPY t0(id, name) FROM STDIN
WITH (
    STREAM_MODE TRUE,
    ON_CONFLICT UPDATE
);

-- Equivalent INSERT INTO statement:
INSERT INTO t0(id, name) VALUES(?, ?)
ON CONFLICT(id) DO UPDATE SET
    id = excluded.id, name = excluded.name;

CREATE TABLE t0 (id int NOT NULL, name text, age int DEFAULT 0, primary key(id));

COPY t0(id, name) FROM STDIN
WITH (
    STREAM_MODE TRUE,
    ON_CONFLICT UPDATE
);

-- Equivalent INSERT INTO statement:
-- For a new row (no matching id), age is set to its default value.
-- For an existing row (matching id), age is not updated.
INSERT INTO t0(id, name, age) VALUES(?, ?, DEFAULT)
ON CONFLICT(id) DO UPDATE SET
    id = excluded.id, name = excluded.name;

-- Create and populate a table
CREATE TABLE copy_to_local (
    id    int,
    age   int,
    name  text
);

INSERT INTO copy_to_local VALUES
    (1, 1, 'a'),
    (1, 2, 'b'),
    (1, 3, 'c'),
    (1, 4, 'd');

-- Export to a local file
\COPY (SELECT * FROM copy_to_local) TO '/root/localfile.txt';

psql -U <username> -p <port> -h <endpoint> -d <databasename> \
    -c "COPY (SELECT * FROM <tablename>) TO STDOUT WITH DELIMITER '|' CSV;" > <filename>

Pré-requisitos

Apenas superusuários e usuários com a função pg_execute_server_program podem executar o hg_dump_to_oss. Conceda a função da seguinte forma:

-- Simple permission model (SPM)
CALL spm_grant('pg_execute_server_program', '<Alibaba Cloud account ID, email address, or RAM user account>');

-- Standard PostgreSQL authorization model
GRANT pg_execute_server_program TO <account>;

Sintaxe

COPY (query) TO PROGRAM 'hg_dump_to_oss
    --AccessKeyId <access_key_id>
    --AccessKeySecret <access_key_secret>
    --Endpoint <oss_classic_network_endpoint>
    --BucketName <bucket_name>
    --DirName <directory>
    [--FileName <file_name>]
    [--BatchSize <n>]'
(DELIMITER ',', HEADER true, FORMAT CSV);

Parâmetros

Exemplos

-- Export from a Hologres internal table to OSS
COPY (SELECT * FROM holo_test LIMIT 2)
TO PROGRAM 'hg_dump_to_oss
    --AccessKeyId <access_id>
    --AccessKeySecret <access_key>
    --Endpoint oss-cn-hangzhou-internal.aliyuncs.com
    --BucketName hologres-demo
    --DirName holotest/
    --FileName file_name
    --BatchSize 3000'
DELIMITER ',';

-- Export from a Hologres foreign table to OSS
COPY (SELECT * FROM foreign_holo_test LIMIT 20)
TO PROGRAM 'hg_dump_to_oss
    --AccessKeyId <access_id>
    --AccessKeySecret <access_key>
    --Endpoint oss-cn-hangzhou-internal.aliyuncs.com
    --BucketName hologres-demo
    --DirName holotest/
    --FileName file_name
    --BatchSize 3000'
(DELIMITER ',', HEADER true);

-- Export to an OSS bucket in a different region
-- (e.g., from a Hologres instance in China (Hangzhou) to an OSS bucket in China (Beijing))
COPY (SELECT * FROM holo_test_1 LIMIT 20)
TO PROGRAM 'hg_dump_to_oss
    --AccessKeyId <access_id>
    --AccessKeySecret <access_key>
    --Endpoint oss-cn-beijing-internal.aliyuncs.com
    --BucketName hologres-demo
    --DirName holotest/
    --FileName file_name
    --BatchSize 3000'
(DELIMITER ',', HEADER true, FORMAT CSV);

Solução de problemas

import org.postgresql.copy.CopyManager;
import org.postgresql.core.BaseConnection;

import java.io.FileOutputStream;
import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.util.Properties;

public class copy_to_local_file {

    public static void main(String args[]) throws Exception {
        System.out.println(copyToFile(getConnection(), "/Users/feng/Workspace/region.tbl", "select * from region"));
    }

    public static Connection getConnection() throws Exception {
        Class.forName("org.postgresql.Driver");
        String url = "jdbc:postgresql://endpoint:port/dbname";
        Properties props = new Properties();
        // Store credentials in environment variables to avoid hardcoding them
        props.setProperty("user", "AAA");     // AccessKey ID
        props.setProperty("password", "BBB"); // AccessKey secret
        return DriverManager.getConnection(url, props);
    }

    /**
     * Streams a Hologres query result into a local file via COPY TO STDOUT.
     *
     * @param connection  Active JDBC connection
     * @param filePath    Destination file path
     * @param SQL_Query   SELECT statement to export
     * @return Path of the written file
     */
    public static String copyToFile(Connection connection, String filePath, String SQL_Query)
            throws SQLException, IOException {

        FileOutputStream fileOutputStream = null;

        try {
            CopyManager copyManager = new CopyManager((BaseConnection) connection);
            fileOutputStream = new FileOutputStream(filePath);
            copyManager.copyOut("COPY (" + SQL_Query + ") TO STDOUT DELIMITER '|' csv", fileOutputStream);
        } finally {
            if (fileOutputStream != null) {
                try {
                    fileOutputStream.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
        return filePath;
    }
}