Columnar table data import

Import data from OSS

DROP SERVER IF EXISTS my_server;
CREATE SERVER my_server FOREIGN DATA WRAPPER oss OPTIONS (
  EXTRA_SERVER_INFO '{"oss_endpoint":"oss-cn-xxxx-internal.aliyuncs.com","oss_bucket":"xxxx","oss_access_key_id":"xxxx","oss_access_key_secret":"xxxx"}'
);
SELECT * FROM mysql.servers;

CREATE TABLE lineitem (
    l_orderkey BIGINT NOT NULL,
    l_partkey BIGINT NOT NULL,
    l_suppkey BIGINT 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 CHAR(1) NOT NULL,
    l_linestatus CHAR(1) NOT NULL,
    l_shipdate DATE NOT NULL,
    l_commitdate DATE NOT NULL,
    l_receiptdate DATE NOT NULL,
    l_shipinstruct CHAR(25) NOT NULL,
    l_shipmode CHAR(10) NOT NULL,
    l_comment VARCHAR(44) NOT NULL,
    PRIMARY KEY (l_orderkey, l_linenumber, l_shipdate),
    ORDER KEY(l_orderkey)
) ENGINE=XENGINE TABLE_FORMAT=COLUMN;

-- Set the parallelism level (optional)
SET xengine_dload_parallel = 2;

-- Use wildcards to import multiple files (OSS mode only)
LOAD DATA OSS INFILE 'my_server/lineitem/lineitem_test*.csv'
INTO TABLE lineitem
FIELDS TERMINATED BY '|';

-- Verify the import results
SELECT * FROM lineitem ORDER BY l_orderkey, l_partkey DESC LIMIT 10;
SELECT COUNT(1) FROM lineitem;

Import data from a local file

Use LOAD DATA LOCAL to import data from a local file.

CREATE TABLE lineitem (
    l_orderkey       BIGINT NOT NULL,
    l_partkey        BIGINT NOT NULL,
    l_suppkey        BIGINT NOT NULL,
    l_linenumber     BIGINT 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     CHAR(1) NOT NULL,
    l_linestatus     CHAR(1) NOT NULL,
    l_shipdate       DATE NOT NULL,
    l_commitdate     DATE NOT NULL,
    l_receiptdate    DATE NOT NULL,
    l_shipinstruct   CHAR(25) NOT NULL,
    l_shipmode       CHAR(10) NOT NULL,
    l_comment        VARCHAR(44) NOT NULL,
    PRIMARY KEY (l_orderkey, l_linenumber, l_shipdate),
    ORDER KEY(l_orderkey)
) ENGINE=XENGINE TABLE_FORMAT=COLUMN;

SET xengine_dload_parallel = 1;

-- Import from a local file (wildcards are not supported)
LOAD DATA LOCAL INFILE '/home/data/lineitem/lineitem_test.csv'
INTO TABLE lineitem
FIELDS TERMINATED BY '|';

-- Verify the import results
SELECT * FROM lineitem ORDER BY l_orderkey, l_partkey DESC LIMIT 10;
SELECT COUNT(1) FROM lineitem;

Parquet syntax

LOAD DATA [LOCAL | OSS] INFILE 'file_name_or_glob'
    [REPLACE | IGNORE]
    FORMAT PARQUET
    INTO TABLE tbl_name
    [PARALLEL n]
    [(col_name [, col_name] ...)]
    [OPTIONS (
        batch_size     = <unsigned int>,
        validate_schema = TRUE | FALSE,
        parallel       = <unsigned int>
    )];

Syntax details

• Auto-detection: If the file name ends with .parquet (case-insensitive), the system automatically enables Parquet import mode, even if you omit the FORMAT PARQUET keyword.

• Parallel loading (PARALLEL n): Specifies the thread count for parallel RowGroup reads. The actual parallelism is the smaller of n and the total number of RowGroups across all files: actual parallelism = min(n, RowGroups x files). If omitted, parallelism defaults to the session variable xengine_dload_parallel.

  • The parallel key in OPTIONS is equivalent to the outer PARALLEL n. When both are set, OPTIONS(parallel=N) takes precedence.

• Selective column import ((col_name, ...)): Lists specific columns to import from the Parquet file. Only listed columns are loaded; unlisted columns receive their default values.

Parquet restrictions

• Data type compatibility: Data types in the Parquet file must be compatible with the target table column types. For example, Parquet STRING maps to VARCHAR or TEXT, but BINARY maps only to BLOB or VARBINARY. See Type mapping for details.

• Schema matching:

  • Positional mode (no column list): The number of columns in the Parquet file must exactly match the number of columns in the target table, and types must be compatible by position.

  • Named mode (column list provided): Each column name in the list must exist in both the Parquet file and the target table, and the types must be compatible.

• Asynchronous index build: Parquet import always uses asynchronous NCI (non-clustered index) build internally. LOAD DATA returns immediately after data is written, and the index builds in the background. During this process, the table supports only columnar queries. You do not need to set xengine_dload_async_build_nci manually.

• Unsupported type: FIXED_SIZE_BINARY Arrow columns are not supported in the current version.

Type mapping

Parquet/Arrow types map to MySQL column types as follows.

Parquet import examples

SET xengine_enable_ctable_hidden_key = ON;
SET xengine_dload_parallel = 4;

CREATE TABLE t_basic (
    id     INT,
    bigval BIGINT,
    amount DOUBLE,
    name   VARCHAR(100)
) ENGINE=XENGINE TABLE_FORMAT=COLUMN;

-- Explicit FORMAT PARQUET
LOAD DATA LOCAL INFILE '/data/test_basic.parquet'
  FORMAT PARQUET INTO TABLE t_basic;

-- Auto-detect by .parquet extension
LOAD DATA LOCAL INFILE '/data/test_basic.parquet' INTO TABLE t_basic;

-- Wait for the asynchronous NCI build to complete before using row plans or DML
SELECT schema_name, table_name, task_status
  FROM information_schema.XENGINE_CTABLE_NCI_BUILD;

-- Import only the id and name columns. Other columns are filled with defaults or NULL.
LOAD DATA LOCAL INFILE '/data/test_basic.parquet'
  FORMAT PARQUET INTO TABLE t_prune (id, name);

-- PARALLEL clause: 4 threads split by RowGroup
LOAD DATA LOCAL INFILE '/data/big.parquet'
  FORMAT PARQUET INTO TABLE t_par PARALLEL 4;

-- Wildcards + PARALLEL: multi-file parallel import
LOAD DATA LOCAL INFILE '/data/multi_*.parquet'
  FORMAT PARQUET INTO TABLE t_multi PARALLEL 8;

LOAD DATA LOCAL INFILE '/data/test_basic.parquet'
  FORMAT PARQUET INTO TABLE t_opts
  OPTIONS(batch_size=65536, validate_schema=TRUE);

-- Single-column primary key
CREATE TABLE t_upk1 (
    id     INT PRIMARY KEY,
    bigval BIGINT,
    amount DOUBLE,
    name   VARCHAR(100)
) ENGINE=XENGINE TABLE_FORMAT=COLUMN;

LOAD DATA LOCAL INFILE '/data/test_basic.parquet'
  FORMAT PARQUET INTO TABLE t_upk1;

-- Composite primary key
CREATE TABLE t_upk_comp (
    id     INT,
    bigval BIGINT,
    amount DOUBLE,
    name   VARCHAR(100),
    PRIMARY KEY (id, bigval)
) ENGINE=XENGINE TABLE_FORMAT=COLUMN;

LOAD DATA LOCAL INFILE '/data/test_basic.parquet'
  FORMAT PARQUET INTO TABLE t_upk_comp;

-- Create the OSS Server (same as the CSV example)
DROP SERVER IF EXISTS my_server;
CREATE SERVER my_server FOREIGN DATA WRAPPER oss OPTIONS (
  EXTRA_SERVER_INFO '{"oss_endpoint": "oss-cn-xxxx.aliyuncs.com",
                      "oss_bucket": "xxxx",
                      "oss_access_key_id": "xxxx",
                      "oss_access_key_secret": "xxxx"}'
);

SET xengine_dload_parallel = 16;

LOAD DATA OSS INFILE 'my_server/path/to/data_*.parquet'
  FORMAT PARQUET INTO TABLE t_basic PARALLEL 16;

Usage

SET xengine_dload_async_build_nci = ON;
SET xengine_dload_parallel = 64;

LOAD DATA OSS INFILE 'my_server/tpch1000g/lineitem.*'
INTO TABLE tpch1000.lineitem
FIELDS TERMINATED BY '|';

Restrictions during NCI build

• DML and DDL operations on the table are blocked during the NCI build.

• Only columnar execution plans are supported. Row-store plans fail with ER_XENGINE_CTABLE_NCI_NOT_READY_FOR_ROWPLAN.

Query examples during asynchronous NCI build

SET use_imci_engine = ON;
SELECT * FROM lineitem ORDER BY l_orderkey, l_partkey DESC LIMIT 10;  -- OK
SELECT COUNT(1) FROM lineitem;  -- OK

SET use_imci_engine = OFF;
SELECT * FROM lineitem ORDER BY l_orderkey, l_partkey DESC LIMIT 10;  -- Error
UPDATE lineitem SET l_partkey=1 WHERE l_orderkey=149999904;  -- Error

Monitor NCI build progress

Query information_schema.XENGINE_CTABLE_NCI_BUILD to monitor progress.

SELECT schema_name, table_name, task_status FROM information_schema.XENGINE_CTABLE_NCI_BUILD ORDER BY subtable_id;

Cancel and rebuild NCI

Cancel an ongoing NCI build with the dbms_ctable package, then rebuild if needed.

-- Cancel the NCI build
CALL dbms_ctable.cancel_nci_build(<subtable_id>);
-- Rebuild the NCI after cancellation
CALL dbms_ctable.rebuild_nci(<subtable_id>);

Create the target tables

This example uses the lineitem table.

SET xengine_enable_ctable_hidden_key = OFF;

DROP DATABASE IF EXISTS tpch100;
CREATE DATABASE tpch100;
USE tpch100;

-- Example: Create the lineitem table
CREATE TABLE lineitem (
    L_ORDERKEY    INTEGER NOT NULL,
    L_PARTKEY     INTEGER NOT NULL,
    L_SUPPKEY     INTEGER NOT NULL,
    L_LINENUMBER  INTEGER 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  CHAR(1) NOT NULL,
    L_LINESTATUS  CHAR(1) NOT NULL,
    L_SHIPDATE    DATE NOT NULL,
    L_COMMITDATE  DATE NOT NULL,
    L_RECEIPTDATE DATE NOT NULL,
    L_SHIPINSTRUCT CHAR(25) NOT NULL,
    L_SHIPMODE     CHAR(10) NOT NULL,
    L_COMMENT      VARCHAR(44) NOT NULL,
    PRIMARY KEY (L_ORDERKEY, L_LINENUMBER, L_SHIPDATE)
) ENGINE=XENGINE TABLE_FORMAT=COLUMN;

Create the OSS Server

DROP SERVER IF EXISTS my_server;

CREATE SERVER my_server
FOREIGN DATA WRAPPER oss OPTIONS
(
  EXTRA_SERVER_INFO '{"oss_endpoint": "oss-cn-xxxx.aliyuncs.com","oss_bucket": "xxxx","oss_access_key_id": "xxxx","oss_access_key_secret": "xxxx"}'
);

Batch import script

The following bash script imports all TPC-H tables in parallel.

#!/usr/bin/env bash
set -u

SOCK=/tmp/mysql29.sock
OUT=output.txt
PREFIX="my_server/imci/tpch100g"
DB=tpch100
PARA=64

: > "$OUT"

tables=(lineitem nation region part supplier partsupp customer orders)

run() {
  local t="$1"
  (
    echo "===== [$t] START $(date '+%F %T') ====="
    sql="SET xengine_dload_parallel=${PARA};
         LOAD DATA OSS INFILE '${PREFIX}/${t}.*'
         INTO TABLE ${DB}.${t}
         FIELDS TERMINATED BY '|';"
    { time mysql -uroot -S "$SOCK" -e "$sql"; } 2>&1
    rc=${PIPESTATUS[0]:-0}
    echo "===== [$t] END   $(date '+%F %T') rc=$rc ====="
    echo
    exit "$rc"
  ) >>"$OUT" 2>&1 &
  echo "[$t] pid=$!"
}

for t in "${tables[@]}"; do
  run "$t"
done

wait
echo "ALL DONE $(date '+%F %T')" >>"$OUT"