PolarDB:Automatically sync data from PolarDB for MySQL to PolarSearch

Data preparation

Run the following SQL statements in PolarDB for MySQL to create a sample database and table, and insert test data.

CREATE DATABASE IF NOT EXISTS db1;
USE db1;
CREATE TABLE IF NOT EXISTS t1 (
    id INT PRIMARY KEY,
    c1 VARCHAR(100),
    c2 VARCHAR(100)
);
INSERT INTO t1(id, c1, c2) VALUES 
(1, 'apple', 'red'),
(2, 'banana', 'yellow'),
(3, 'grape', 'purple');

Create a synchronization link

Use the dbms_etl.sync_by_table stored procedure to create a synchronization task from the db1.t1 table to the dest index on the PolarSearch node.

Syntax

call dbms_etl.sync_by_table("search", "<source_table>", "<sink_table>", "<column_list>");

Parameter description

Limits

• The source table must have a primary key or unique key.

• You cannot use the same source table or destination table in different synchronization links.

• After creating a link, newly added columns in the source table are not synchronized by default. To synchronize new columns, rebuild the link.

• If you want to use a custom index configuration, manually create the index and define its settings in the PolarSearch node before creating the synchronization link. If the destination index does not exist when the link is created, the system creates it automatically.

Examples

• Synchronize the entire db1.t1 table to the dest index in PolarSearch:

  call dbms_etl.sync_by_table("search", "db1.t1", "dest", "");

• Synchronize only the c1 and c2 columns from the db1.t1 table to the dest index:

  call dbms_etl.sync_by_table("search", "db1.t1", "dest", "c1, c2");

Verify data

Connect to the PolarSearch node and use an Elasticsearch-compatible REST API to query and confirm that data has been synchronized.

# Replace <polarsearch_endpoint> with the PolarSearch node endpoint
curl -u <user>:<password> -X GET "http://<polarsearch_endpoint>/dest/_search"

Data preparation

Run the following SQL statements in PolarDB for MySQL to create sample databases and tables, and insert test data.

CREATE DATABASE IF NOT EXISTS db1;
CREATE DATABASE IF NOT EXISTS db2;
CREATE DATABASE IF NOT EXISTS db3;

CREATE TABLE IF NOT EXISTS db1.t1 (id INT PRIMARY KEY, c1 INT);
CREATE TABLE IF NOT EXISTS db2.t2 (id INT PRIMARY KEY, c2 INT);
CREATE TABLE IF NOT EXISTS db3.t3 (id INT PRIMARY KEY, c3 VARCHAR(10));

INSERT INTO db1.t1(id, c1) VALUES (1, 11), (2, 22), (3, 33);
INSERT INTO db2.t2(id, c2) VALUES (1, 111), (2, 222), (4, 444);
INSERT INTO db3.t3(id, c3) VALUES (1, 'aaa'), (3, 'ccc'), (4, 'ddd');

Create a synchronization link

Use the dbms_etl.sync_by_map stored procedure to join data from multiple tables and aggregate it into an index on a PolarSearch node.

Syntax

call dbms_etl.sync_by_map(
    "search",
    "<columns_map>", -- Mapping between destination index fields and source table fields
    "<join_fields>", -- Join keys between tables
    "<join_types>",  -- Join types (inner, left)
    "<filter>"       -- Data filter condition
);

Parameter description

Limits

• All source tables involved in the synchronization must have primary keys.

• This feature uses stream processing and guarantees only eventual consistency during synchronization.

• The update mode for the destination index is delete-then-insert. If you do not want queries to access intermediate states of deleted data, set the session variable before running the command: set sink_options = "'ignore-delete' = 'true'"; to ignore data deletion on the PolarSearch node.

Examples

• INNER JOIN two tables: Join db1.t1 and db2.t2 on the id field, and synchronize t1.c1 and t2.c2 to fields c1 and c2 in the dest index.

  call dbms_etl.sync_by_map(
    "search",
    "dest.id(db1.t1.id),dest.c1(db1.t1.c1),dest.c2(db2.t2.c2)",
    "dest.id=db1.t1.id,db2.t2.id", 
    "inner",
     ""
  );

• Mixed JOIN and filter across multiple tables: Join three tables—db1.t1, db2.t2, and db3.t3—where t1 and t2 use INNER JOIN, and t1 and t3 use LEFT JOIN. Filter data where t1.c1 > 10 and t2.c2 < 100.

  call dbms_etl.sync_by_map(
    "search", 
    "dest.id(db1.t1.id),dest.c1(db1.t1.c1),dest.c2(db2.t2.c2),dest.c3(db3.t3.c3)", 
    "dest.id=db1.t1.id,db2.t2.id,db3.t3.id", 
    "inner,left", 
    "db1.t1.c1 > 10 and db2.t2.c2 < 100"
  );

Verify data

Connect to the PolarSearch node and use an Elasticsearch-compatible REST API to query and confirm that data has been synchronized.

# Replace <polarsearch_endpoint> with the PolarSearch node endpoint
curl -u <user>:<password> -X GET "http://<polarsearch_endpoint>/dest/_search"

Create a synchronization link

For scenarios requiring complex transformations, aggregations, or calculations, the dbms_etl.sync_by_sql stored procedure supports defining data synchronization logic using Flink SQL syntax.

Syntax

call dbms_etl.sync_by_sql("search", "<sync_sql>");

Example

The system automatically replaces placeholders in the SQL: {mysql_host}, {mysql_port}, {mysql_user}, {mysql_password}, {search_host}, {search_port}, {search_user}, and {search_password}. Write your SQL using these fixed placeholders.

CALL dbms_etl.sync_by_sql("search", "
-- Step 1: Define PolarDB source table
CREATE TEMPORARY TABLE `db1`.`sbtest1` (
  `id`   BIGINT,
  `k`    BIGINT,
  `c`    STRING,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector' = 'mysql',
  'hostname' = '{mysql_host}',
  'port' = '{mysql_port}',
  'username' = '{mysql_user}',       -- Never use plaintext in production
  'password' = '{mysql_password}',   -- Never use plaintext in production
  'database-name' = 'db1',
  'table-name' = 'sbtest1'
);

-- Step 2: Define PolarSearch destination table
CREATE TEMPORARY TABLE `dest` (
  `k`  BIGINT,
  `max_c` STRING,
  PRIMARY KEY (`k`) NOT ENFORCED
) WITH (
  'connector' = 'opensearch',
  'hosts' = '{search_host}:{search_port}',   
  'index' = 'dest',
  'username' = '{search_user}',     -- Never use plaintext in production
  'password' = '{search_password}'  -- Never use plaintext in production
);

-- Step 3: Define computation and insert logic
INSERT INTO `dest`
SELECT
    `t1`.`k`,
    MAX(`t1`.`c`)
FROM `db1`.`sbtest1` AS `t1`
GROUP BY `t1`.`k`;
");

Verify data

Connect to the PolarSearch node and use an Elasticsearch-compatible REST API to query and confirm that data has been synchronized.

# Replace <polarsearch_endpoint> with the PolarSearch node endpoint
curl -u <user>:<password> -X GET "http://<polarsearch_endpoint>/dest/_search"