Hologres:COPY

Parameter

Nilai ON_CONFLICT:

Atomicitas

COPY standar (STREAM_MODE FALSE) menjamin atomicitas: seluruh operasi berhasil atau dikembalikan (rollback).

Mode copy tetap (STREAM_MODE TRUE) menggunakan lock tingkat baris alih-alih lock tingkat tabel, sehingga atomicitas tidak dijamin. Jika suatu baris berisi data tidak valid, error dilaporkan hanya untuk baris tersebut—baris lainnya mungkin sebagian tertulis atau tidak tertulis sama sekali.

Query log

Di Hologres V3.0 dan versi lebih baru, setiap operasi COPY menghasilkan dua catatan di hologres.hg_query_log: satu untuk perintah COPY dan satu untuk INSERT yang dijalankannya secara internal. Hubungkan keduanya menggunakan ID transaksi:

SELECT
    query_id,
    query,
    extended_info
FROM
    hologres.hg_query_log
WHERE
    extended_info ->> 'source_trx' = '<transaction_id>'  -- Dapatkan ID transaksi dari field trans_id dalam catatan log COPY
ORDER BY
    query_start;

Pada versi sebelum V3.0, setiap operasi COPY menghasilkan satu catatan saja.

Impor dari PostgreSQL client (stdin)

PostgreSQL client hanya dapat membaca dari stdin. Konsol HoloWeb tidak mendukung impor stdin.

Contoh 1: Impor teks berpemisah

-- Buat tabel target
CREATE TABLE copy_test (
    id    int,
    age   int,
    name  text
);

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

-- Verifikasi
SELECT * FROM copy_test;

Contoh 2: Impor file CSV

-- Buat tabel target
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
);

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

-- Verifikasi
SELECT * FROM partsupp;

Contoh 3: Impor file lokal menggunakan psql

Arahkan file lokal ke stdin dengan operator redirect shell psql:

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

Contoh berikut mengimpor file lokal copy_test menggunakan perintah ini:

File berisi:

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

Setelah impor, kueri hasilnya di psql:

Impor dari klien JDBC menggunakan CopyManager

Klien Java Database Connectivity (JDBC) dapat menggunakan CopyManager—wrapper API driver JDBC PostgreSQL untuk COPY—untuk mengalirkan file ke 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();
        // Simpan kredensial dalam variabel lingkungan untuk menghindari hardcoding
        props.setProperty("user", "AAA");     // ID AccessKey
        props.setProperty("password", "BBB"); // Rahasia AccessKey
        return DriverManager.getConnection(url, props);
    }

    /**
     * Mengalirkan file lokal ke Hologres melalui COPY FROM STDIN.
     *
     * @param connection  Koneksi JDBC aktif
     * @param filePath    Path ke file lokal
     * @param tableName   Tabel Hologres target
     * @return Jumlah baris yang diimpor
     */
    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;
    }
}

Mode copy tetap

Mode copy tetap (STREAM_MODE TRUE) menggunakan rencana eksekusi tetap yang telah dikompilasi sebelumnya untuk mempercepat impor COPY berulang. Ini adalah optimisasi khusus Hologres yang tersedia sejak V1.3.17 dan hanya berlaku untuk impor. Untuk perbandingan dengan mode penulisan batch lainnya, lihat Perbandingan mode penulisan batch. Untuk mekanisme dasarnya, lihat Percepat eksekusi SQL dengan rencana tetap.

Menulis ke subset kolom — pembaruan parsial

Ketika ON_CONFLICT UPDATE diatur dan COPY hanya menulis ke beberapa kolom, kolom yang tidak termasuk dalam daftar COPY tidak dimodifikasi:

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

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

Menulis ke subset kolom — kolom dengan nilai default

Ketika kolom yang tidak termasuk dalam daftar COPY memiliki nilai DEFAULT, Hologres menerapkan nilai default hanya untuk baris baru. Baris yang sudah ada dan cocok berdasarkan primary key tidak diperbarui untuk kolom tersebut:

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

-- Pernyataan INSERT INTO yang setara:
-- Untuk baris baru (tidak ada id yang cocok), age diatur ke nilai default-nya.
-- Untuk baris yang sudah ada (id yang cocok), age tidak diperbarui.
INSERT INTO t0(id, name, age) VALUES(?, ?, DEFAULT)
ON CONFLICT(id) DO UPDATE SET
    id = excluded.id, name = excluded.name;

Ekspor ke file lokal

Kedua metode berikut hanya tersedia pada PostgreSQL client.

Menggunakan meta-perintah `\copy` (psql)

-- Buat dan isi tabel
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');

-- Ekspor ke file lokal
\COPY (SELECT * FROM copy_to_local) TO '/root/localfile.txt';

Menggunakan pengalihan stdout (psql)

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

Ekspor ke Object Storage Service (OSS)

Gunakan program hg_dump_to_oss dengan COPY TO PROGRAM untuk mengekspor data Hologres ke bucket OSS. Setiap ekspor dibatasi hingga 5 GB.

-- Model izin sederhana (SPM)
CALL spm_grant('pg_execute_server_program', '<ID akun Alibaba Cloud, alamat email, atau akun Pengguna RAM>');

-- Model otorisasi PostgreSQL standar
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);

-- Ekspor dari tabel internal Hologres ke 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 ',';

-- Ekspor dari tabel eksternal Hologres ke 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);

-- Ekspor ke bucket OSS di wilayah berbeda
-- (misalnya, dari instans Hologres di Tiongkok (Hangzhou) ke bucket OSS di Tiongkok (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);

Ekspor dari klien JDBC menggunakan 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();
        // Simpan kredensial dalam variabel lingkungan untuk menghindari hardcoding
        props.setProperty("user", "AAA");     // ID AccessKey
        props.setProperty("password", "BBB"); // Rahasia AccessKey
        return DriverManager.getConnection(url, props);
    }

    /**
     * Mengalirkan hasil kueri Hologres ke file lokal melalui COPY TO STDOUT.
     *
     * @param connection  Koneksi JDBC aktif
     * @param filePath    Path file tujuan
     * @param SQL_Query   Pernyataan SELECT untuk diekspor
     * @return Path file yang ditulis
     */
    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;
    }
}