Hologres:COPY

パラメーター

ON_CONFLICT の値：

原子性

標準の COPY (STREAM_MODE FALSE) は原子性を保証します：操作全体が成功するか、ロールバックされます。

固定コピーモード (STREAM_MODE TRUE) は、テーブルレベルロックの代わりに行レベルロックを使用するため、原子性は保証されません。行に無効なデータが含まれている場合、その行に対してのみエラーが報告され、残りの行は部分的に書き込まれるか、まったく書き込まれない可能性があります。

クエリログ

Hologres V3.0 以降では、各 COPY 操作は hologres.hg_query_log に 2 つのレコードを生成します：1 つは COPY コマンド用、もう 1 つは内部で実行される INSERT 用です。トランザクション ID を使用してこれらをリンクします：

SELECT
    query_id,
    query,
    extended_info
FROM
    hologres.hg_query_log
WHERE
    extended_info ->> 'source_trx' = '<transaction_id>'  -- COPY ログレコードの trans_id フィールドからトランザクション ID を取得します
ORDER BY
    query_start;

V3.0 より前のバージョンでは、各 COPY 操作は単一のレコードを生成します。

PostgreSQL クライアント (stdin) からのインポート

PostgreSQL クライアントは stdin からのみ読み取ることができます。HoloWeb コンソールは stdin インポートをサポートしていません。

例 1：区切り文字付きテキストのインポート

-- ターゲットテーブルを作成
CREATE TABLE copy_test (
    id    int,
    age   int,
    name  text
);

-- stdin からデータをインポート
COPY copy_test FROM STDIN WITH DELIMITER AS ',' NULL AS '';
53444,24,wangming
55444,38,ligang
55444,38,luyong
\.

-- 検証
SELECT * FROM copy_test;

例 2：CSVファイルのインポート

-- ターゲットテーブルを作成
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
);

-- stdin から CSV をインポート
COPY partsupp FROM STDIN WITH DELIMITER '|' CSV;
1|2|3325|771.64|final theodolites
1|25002|8076|993.49|ven ideas
\.

-- 検証
SELECT * FROM partsupp;

例 3：psql を使用したローカルファイルのインポート

psql シェルのリダイレクト演算子を使用して、ローカルファイルを stdin にリダイレクトします：

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

次の例では、このコマンドを使用してローカルファイル copy_test をインポートします：

ファイルの内容は次のとおりです：

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

インポート後、psql で結果をクエリします：

CopyManager を使用した JDBC クライアントからのインポート

Java Database Connectivity (JDBC) クライアントは、CopyManager (COPY 用の PostgreSQL JDBC ドライバーの API ラッパー) を使用して、ファイルを 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();
        // 認証情報をハードコーディングしないように、環境変数に保存します
        props.setProperty("user", "AAA");     // AccessKey ID
        props.setProperty("password", "BBB"); // AccessKey Secret
        return DriverManager.getConnection(url, props);
    }

    /**
     * COPY FROM STDIN を使用してローカルファイルを Hologres にストリーミングします。
     *
     * @param connection  アクティブな JDBC 接続
     * @param filePath    ローカルファイルのパス
     * @param tableName   ターゲットHologresテーブル
     * @return インポートされた行数
     */
    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;
    }
}

固定コピーモード

固定コピーモード (STREAM_MODE TRUE) は、プリコンパイルされた固定実行計画を使用して、繰り返し行われる COPY インポートを高速化します。これは、V1.3.17 から利用可能な Hologres 固有の最適化であり、インポートにのみ適用されます。他のバッチ書き込みモードとの比較については、「バッチ書き込みモードの比較」をご参照ください。内部的な仕組みについては、「固定計画によるSQL実行の高速化」をご参照ください。

列のサブセットへの書き込み — 部分更新

ON_CONFLICT UPDATE が設定されている場合で、COPY が一部の列にのみ書き込む場合、COPY リストに含まれていない列は変更されません：

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

-- 同等の INSERT INTO 文：
INSERT INTO t0(id, name) VALUES(?, ?)
ON CONFLICT(id) DO UPDATE SET
    id = excluded.id, name = excluded.name;

列のサブセットへの書き込み — デフォルト値を持つ列

COPY リストに含まれていない列に DEFAULT 値がある場合、Hologres は新規の行にのみデフォルト値を適用します。プライマリーキーで一致する既存の行は、その列については更新されません：

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

-- 同等の INSERT INTO 文：
-- 新しい行 (一致する id がない) の場合、age はデフォルト値に設定されます。
-- 既存の行 (一致する id がある) の場合、age は更新されません。
INSERT INTO t0(id, name, age) VALUES(?, ?, DEFAULT)
ON CONFLICT(id) DO UPDATE SET
    id = excluded.id, name = excluded.name;

ローカルファイルへのエクスポート

以下の両方の方法は、PostgreSQL クライアントでのみ利用可能です。

\copy メタコマンドの使用 (psql)

-- テーブルを作成してデータを入力
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');

-- ローカルファイルにエクスポート
\COPY (SELECT * FROM copy_to_local) TO '/root/localfile.txt';

stdout リダイレクションの使用 (psql)

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

Object Storage Service (OSS) へのエクスポート

hg_dump_to_oss プログラムと COPY TO PROGRAM を使用して、Hologres データを OSS バケットにエクスポートします。各エクスポートは 5 GB に制限されます。

-- 簡易権限モデル (SPM)
CALL spm_grant('pg_execute_server_program', '<Alibaba Cloud account ID, email address, or RAM user account>');

-- 標準 PostgreSQL 権限モデル
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);

-- Hologres 内部テーブルから 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 ',';

-- Hologres 外部テーブルから 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);

-- 異なるリージョンの OSS バケットへエクスポート
-- (例：中国 (杭州) の Hologres インスタンスから中国 (北京) の OSS バケットへ)
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);

CopyManager を使用した JDBC クライアントからのエクスポート

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();
        // 認証情報をハードコーディングしないように、環境変数に保存します
        props.setProperty("user", "AAA");     // AccessKey ID
        props.setProperty("password", "BBB"); // AccessKey Secret
        return DriverManager.getConnection(url, props);
    }

    /**
     * COPY TO STDOUT を使用して Hologres クエリの結果をローカルファイルにストリーミングします。
     *
     * @param connection  アクティブな JDBC 接続
     * @param filePath    宛先ファイルパス
     * @param SQL_Query   エクスポートする SELECT 文
     * @return 書き込まれたファイルのパス
     */
    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;
    }
}