Hologres:COPY

Parâmetros

Valores de ON_CONFLICT:

Atomicidade

A operação COPY padrão (STREAM_MODE FALSE) garante atomicidade: toda a operação é concluída com sucesso ou é revertida.

O modo de cópia fixa (STREAM_MODE TRUE) usa um bloqueio no nível de linha em vez de um bloqueio no nível de tabela; portanto, não há garantia de atomicidade. Se uma linha contiver dados inválidos, o sistema reporta um erro apenas para essa linha — as linhas restantes podem ser gravadas parcialmente ou ignoradas.

Log de consultas

A partir do Hologres V3.0, cada operação COPY gera dois registros em hologres.hg_query_log: um para o comando COPY e outro para o INSERT executado internamente. Vincule-os pelo ID da transaçã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;

Em versões anteriores à V3.0, cada operação COPY gera um único registro.

Importar de um cliente PostgreSQL (stdin)

O cliente PostgreSQL lê apenas do stdin. O console do HoloWeb não suporta importações via stdin.

Exemplo 1: Importar texto delimitado

-- 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;

Exemplo 2: Importar um arquivo CSV

-- 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;

Exemplo 3: Importar um arquivo local usando psql

Redirecione um arquivo local para o stdin usando o operador de redirecionamento do shell psql:

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

O exemplo a seguir importa o arquivo local copy_test com este comando:

O arquivo contém:

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

Após a importação, consulte os resultados no psql:

Importar de um cliente JDBC usando CopyManager

Os clientes Java Database Connectivity (JDBC) podem usar o CopyManager — wrapper de API do driver JDBC do PostgreSQL para COPY — para transmitir arquivos ao Hologres.

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;
    }
}

Modo de cópia fixa

O modo de cópia fixa (STREAM_MODE TRUE) usa planos de execução fixos pré-compilados para acelerar importações COPY repetidas. Essa é uma otimização específica do Hologres, disponível desde a V1.3.17, e aplica-se apenas a importações. Para comparar com outros modos de gravação em lote, consulte Comparison of batch write modes. Para entender o mecanismo subjacente, consulte Accelerate SQL execution with fixed plan.

Gravar em um subconjunto de colunas — atualização parcial

Quando ON_CONFLICT UPDATE está definido e o COPY grava apenas em algumas colunas, as colunas não incluídas na lista do COPY não sofrem modificação:

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;

Gravar em um subconjunto de colunas — colunas com valores padrão

Quando uma coluna não incluída na lista do COPY possui um valor DEFAULT, o Hologres aplica o valor padrão apenas a novas linhas. O sistema não atualiza as linhas existentes correspondentes à chave primária nessa coluna:

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;

Exportar para um arquivo local

Os dois métodos abaixo estão disponíveis apenas no cliente PostgreSQL.

Usar o metacomando \copy (psql)

-- 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';

Usar redirecionamento de stdout (psql)

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

Exportar para o Object Storage Service (OSS)

Use o programa hg_dump_to_oss com COPY TO PROGRAM para exportar dados do Hologres para um bucket do OSS. O limite de cada exportação é de 5 GB.

-- 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>;

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);

-- 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);

Exportar de um cliente JDBC usando CopyManager

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;
    }
}