Hologres:Migrate data from ClickHouse

clickhouse-client --host="<host>" --port="<port>" --user="<username>" --password="<password>" \
  --query="SHOW databases" > database.list

clickhouse-client --host="<host>" --port="<port>" --user="<username>" --password="<password>" \
  --query="SHOW tables from <database_name>" > table.list

SELECT DISTINCT database, name FROM system.tables WHERE database != 'system';

clickhouse-client --host="<host>" --port="<port>" --user="<username>" --password="<password>" \
  --query="SHOW CREATE TABLE <database_name>.<table_name>" > table.sql

SELECT * FROM system.tables
WHERE database = '<database_name>' AND engine != 'Distributed';

-- lineitem on ClickHouse
CREATE TABLE lineitem_local ON CLUSTER default(
  l_orderkey            UInt64,
  l_partkey             UInt32,
  l_suppkey             UInt32,
  l_linenumber          UInt32,
  l_quantity            decimal(15,2),
  l_extendedprice       decimal(15,2),
  l_discount            decimal(15,2),
  l_tax                 decimal(15,2),
  l_returnflag          LowCardinality(String),
  l_linestatus          LowCardinality(String),
  l_shipdate            Date,
  l_commitdate          Date,
  l_receiptdate         Date,
  l_shipinstruct        LowCardinality(String),
  l_shipmode            LowCardinality(String),
  l_comment             LowCardinality(String)
) ENGINE = MergeTree
PARTITION BY toYear(l_shipdate)
ORDER BY (l_orderkey, l_linenumber);

CREATE TABLE lineitem ON CLUSTER default AS lineitem_local
ENGINE = Distributed(default, default, lineitem_local, l_orderkey);

-- lineitem on Hologres
-- Create a table group with 32 shards
CALL hg_create_table_group ('lineitem_tg', 32);
BEGIN;
CREATE TABLE LINEITEM (
    L_ORDERKEY      BIGINT        NOT NULL,
    L_PARTKEY       INT           NOT NULL,
    L_SUPPKEY       INT           NOT NULL,
    L_LINENUMBER    INT           NOT NULL,
    L_QUANTITY      DECIMAL(15,2) NOT NULL,
    L_EXTENDEDPRICE DECIMAL(15,2) NOT NULL,
    L_DISCOUNT      DECIMAL(15,2) NOT NULL,
    L_TAX           DECIMAL(15,2) NOT NULL,
    L_RETURNFLAG    TEXT          NOT NULL,
    L_LINESTATUS    TEXT          NOT NULL,
    L_SHIPDATE      TIMESTAMPTZ   NOT NULL,
    L_COMMITDATE    TIMESTAMPTZ   NOT NULL,
    L_RECEIPTDATE   TIMESTAMPTZ   NOT NULL,
    L_SHIPINSTRUCT  TEXT          NOT NULL,
    L_SHIPMODE      TEXT          NOT NULL,
    L_COMMENT       TEXT          NOT NULL,
    PRIMARY KEY (L_ORDERKEY, L_LINENUMBER)
);
-- sorting_key (l_orderkey, l_linenumber) maps to clustering key and segment key
CALL set_table_property('LINEITEM', 'clustering_key', 'L_SHIPDATE,L_ORDERKEY');
CALL set_table_property('LINEITEM', 'segment_key', 'L_SHIPDATE');
CALL set_table_property('LINEITEM', 'table_group', 'lineitem_tg');
CALL set_table_property('LINEITEM', 'distribution_key', 'L_ORDERKEY');
-- LowCardinality columns: enable bitmap and dictionary encoding
CALL set_table_property('LINEITEM', 'bitmap_columns', 'L_RETURNFLAG,L_LINESTATUS,L_SHIPINSTRUCT,L_SHIPMODE,L_COMMENT');
CALL set_table_property('LINEITEM', 'dictionary_encoding_columns', 'L_RETURNFLAG,L_LINESTATUS,L_SHIPINSTRUCT,L_SHIPMODE,L_COMMENT');
CALL set_table_property('LINEITEM', 'time_to_live_in_seconds', '31536000');
COMMIT;

-- customer on ClickHouse
CREATE TABLE customer_local ON CLUSTER default(
  c_custkey     UInt32,
  c_name        String,
  c_address     String,
  c_nationkey   UInt32,
  c_phone       LowCardinality(String),
  c_acctbal     decimal(15,2),
  c_mktsegment  LowCardinality(String),
  c_comment     LowCardinality(String)
) ENGINE = MergeTree
ORDER BY (c_custkey);

CREATE TABLE customer ON CLUSTER default AS customer_local
ENGINE = Distributed(default, default, customer_local, c_custkey);

-- customer on Hologres
BEGIN;
CREATE TABLE CUSTOMER (
    C_CUSTKEY    INT           NOT NULL PRIMARY KEY,
    C_NAME       TEXT          NOT NULL,
    C_ADDRESS    TEXT          NOT NULL,
    C_NATIONKEY  INT           NOT NULL,
    C_PHONE      TEXT          NOT NULL,
    C_ACCTBAL    DECIMAL(15,2) NOT NULL,
    C_MKTSEGMENT TEXT          NOT NULL,
    C_COMMENT    TEXT          NOT NULL
);
CALL set_table_property('CUSTOMER', 'distribution_key', 'C_CUSTKEY');
CALL set_table_property('CUSTOMER', 'table_group', 'lineitem_tg');
CALL set_table_property('CUSTOMER', 'bitmap_columns', 'C_CUSTKEY,C_NATIONKEY,C_NAME,C_ADDRESS,C_PHONE,C_MKTSEGMENT,C_COMMENT');
CALL set_table_property('CUSTOMER', 'dictionary_encoding_columns', 'C_NAME,C_ADDRESS,C_PHONE,C_MKTSEGMENT,C_COMMENT');
CALL set_table_property('CUSTOMER', 'time_to_live_in_seconds', '31536000');
COMMIT;

PGUSER="<username>" PGPASSWORD="<password>" psql \
  -h "<host>" -p "<port>" -d "<database_name>" -f table.sql

clickhouse-client --host="<host>" --port="<port>" --user="<username>" --password="<password>" \
  --query="SELECT * FROM <database_name>.<table_name> FORMAT CSV" > table.csv

PGUSER="<username>" PGPASSWORD="<password>" psql \
  -h "<host>" -p "<port>" -d "<database_name>" \
  -c "COPY <schema_name>.<table_name> FROM STDIN (FORMAT 'csv')" < table.csv

SELECT COUNT(*) FROM <schema_name>.<table_name>;

SELECT
  l_returnflag,
  l_linestatus,
  sum(l_quantity) AS sum_qty,
  sum(l_extendedprice) AS sum_base_price,
  sum(l_extendedprice * (1 - l_discount)) AS sum_disc_price,
  sum(l_extendedprice * (1 - l_discount) * (1 + l_tax)) AS sum_charge,
  avg(l_quantity) AS avg_qty,
  avg(l_extendedprice) AS avg_price,
  avg(l_discount) AS avg_disc,
  count(*) AS count_order
FROM lineitem
WHERE l_shipdate <= date '1998-12-01' - interval '90' day
GROUP BY l_returnflag, l_linestatus
ORDER BY l_returnflag, l_linestatus;

SELECT
  l_returnflag,
  l_linestatus,
  sum(l_quantity) AS sum_qty,
  sum(l_extendedprice) AS sum_base_price,
  sum(l_extendedprice * (1 - l_discount)) AS sum_disc_price,
  sum(l_extendedprice * (1 - l_discount) * (1 + l_tax)) AS sum_charge,
  avg(l_quantity) AS avg_qty,
  avg(l_extendedprice) AS avg_price,
  avg(l_discount) AS avg_disc,
  count(*) AS count_order
FROM lineitem
WHERE l_shipdate <= date '1998-12-01' - interval '90' day
GROUP BY l_returnflag, l_linestatus
ORDER BY l_returnflag, l_linestatus;

SELECT
  o_orderpriority,
  count(*) AS order_count
FROM orders
WHERE o_orderdate >= date '1993-07-01'
  AND o_orderdate < date '1993-07-01' + interval '3' month
  AND o_orderdate IN (
    SELECT o_orderdate
    FROM lineitem, orders
    WHERE l_orderkey = o_orderkey
      AND l_commitdate < l_receiptdate
  )
GROUP BY o_orderpriority
ORDER BY o_orderpriority;

SELECT
  o_orderpriority,
  count(*) AS order_count
FROM orders
WHERE o_orderdate >= date '1993-07-01'
  AND o_orderdate < date '1993-07-01' + interval '3' month
  AND EXISTS (
    SELECT *
    FROM lineitem
    WHERE l_orderkey = o_orderkey
      AND l_commitdate < l_receiptdate
  )
GROUP BY o_orderpriority
ORDER BY o_orderpriority;

SELECT
  ps_partkey,
  sum(ps_supplycost * ps_availqty) AS value
FROM partsupp, supplier, nation
WHERE ps_suppkey = s_suppkey
  AND s_nationkey = n_nationkey
  AND n_name = 'GERMANY'
GROUP BY ps_partkey
HAVING sum(ps_supplycost * ps_availqty) > (
  SELECT sum(ps_supplycost * ps_availqty) * toDecimal32(0.0000010000, 9)
  FROM partsupp, supplier, nation
  WHERE ps_suppkey = s_suppkey
    AND s_nationkey = n_nationkey
    AND n_name = 'GERMANY'
)
ORDER BY value DESC
LIMIT 100;

SELECT
  ps_partkey,
  sum(ps_supplycost * ps_availqty) AS value
FROM partsupp, supplier, nation
WHERE ps_suppkey = s_suppkey
  AND s_nationkey = n_nationkey
  AND n_name = 'GERMANY'
GROUP BY ps_partkey
HAVING sum(ps_supplycost * ps_availqty) > (
  SELECT sum(ps_supplycost * ps_availqty) * 0.0000010000
  FROM partsupp, supplier, nation
  WHERE ps_suppkey = s_suppkey
    AND s_nationkey = n_nationkey
    AND n_name = 'GERMANY'
)
ORDER BY value DESC
LIMIT 100;