PolarDB:Data synchronization from PolarDB MySQL to PolarSearch

Create a search view

Syntax

CREATE SEARCH VIEW view_name [(column_list, PRIMARY KEY (pk_column_list))] AS select_statement;

Parameters

Usage notes

• To use a custom target index, create and configure it on the PolarSearch node before creating the search view. If the target index does not exist at creation time, the system creates it automatically.

• To configure advanced synchronization parameters—such as JSON field conversions or routing fields for multi-table aggregation—see AutoETL Parameter Configuration and Practice Cases.

Data preparation

The following examples use test tables. Run these statements in PolarDB for MySQL to set up the data:

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

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

USE db2;
CREATE TABLE IF NOT EXISTS t2 (id INT PRIMARY KEY, c2 INT);
INSERT INTO t2(id, c2) VALUES (1, 111), (2, 222), (4, 444);

Examples

• Full table sync — synchronize all columns from db1.t1. The view name view_test becomes the target index name in PolarSearch.

  CREATE SEARCH VIEW view_test AS SELECT * FROM db1.t1;

• Column selection — synchronize only c1 and c2.

  CREATE SEARCH VIEW view_test1 AS SELECT c1, c2 FROM db1.t1;

• Conditional filtering — synchronize only rows where c1 > 10.

  CREATE SEARCH VIEW view_test2 AS SELECT id, c1, c2 FROM db1.t1 WHERE c1 > 10;

• Multi-table JOIN — join db1.t1 and db2.t2 on id and synchronize the result.

  CREATE SEARCH VIEW view_test3(id, c1, c2) AS SELECT t1.id, t1.c1, t2.c2 FROM db1.t1 AS t1 LEFT JOIN db2.t2 AS t2 ON t1.id = t2.id;

• Multi-table UNION — merge two tables with compatible structures. Both SELECT statements must return the same number and types of columns.

  CREATE SEARCH VIEW view_test4(id, c2) AS SELECT id, c2 FROM db1.t1 UNION ALL SELECT id, c2 FROM db2.t2;

• Grouped aggregation — group and aggregate before synchronization. When using GROUP BY, define the columns and primary key manually.

  CREATE SEARCH VIEW view_test5 (id, max_c) AS SELECT t1.id, MAX(t1.c1) AS max_c FROM db1.t1 GROUP BY t1.id;

Verify data synchronization

Check the synchronization status:

SHOW SEARCH VIEW STATUS;

A status of active means the view is synchronizing normally:

+------------+--------+----------+---------+---------------------+---------------------+
| View Name  | Type   | Status   | Message | Created_at          | Updated_at          |
+------------+--------+----------+---------+---------------------+---------------------+
| view_test  | search | active   |         | 2026-03-18 18:44:12 | 2026-03-18 18:51:37 |
+------------+--------+----------+---------+---------------------+---------------------+

Once the status is active, query the synchronized data through the PolarSearch node using the Elasticsearch-compatible REST API:

# Replace <user>:<password> with the PolarSearch node credentials
# and <polarsearch_endpoint> with the endpoint and port of the PolarSearch node.
curl -u <user>:<password> -X GET "http://<polarsearch_endpoint>/view_test/_search"

The response includes the synchronized documents:

{
  "hits": {
    "total": { "value": 3, "relation": "eq" },
    "hits": [
      { "_id": "1", "_source": { "id": 1, "c1": "apple", "c2": "red" } },
      { "_id": "2", "_source": { "id": 2, "c1": "banana", "c2": "yellow" } },
      { "_id": "3", "_source": { "id": 3, "c1": "grape", "c2": "purple" } }
    ]
  }
}

Manage search views

• View all search views and their status:

  SHOW SEARCH VIEW STATUS;

• View the definition of a specific search view:

  SHOW CREATE SEARCH VIEW view_test;

  Output:

  +-----------+------------------------------------------------------+
  | View Name | Create Search View                                   |
  +-----------+------------------------------------------------------+
  | view_test | CREATE SEARCH VIEW view_test AS SELECT * FROM db1.t1 |
  +-----------+------------------------------------------------------+

Modify a search view

To change synchronization logic—such as adding a new field or updating a query condition—use a "new index + new search view" approach to avoid affecting running queries.

1. Create a new search view that synchronizes data to a new PolarSearch index.

2. Run SHOW SEARCH VIEW STATUS and wait until the synchronization latency for the new view drops to 0–1 seconds.

3. Switch your business queries from the old index to the new one.

4. Drop the old search view.

For details on how DDL changes to a source table affect a search view, see DDL Change Rules and Practices.

Drop a search view

DROP SEARCH VIEW view_name;

State transitions when dropping:

• `active`: Changes to dropping first, then to dropped after resources are cleaned up.

• `dropped`: All information is permanently deleted.

• Other states: Dropping is not supported.

Create a synchronization link

Syntax

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

The system automatically substitutes the following placeholders in <sync_sql>. You only need to write the SQL using these fixed placeholders—no manual connection configuration required.

Example

The following example defines a source table, a target table, and a GROUP BY aggregation:

CALL dbms_etl.sync_by_sql("search", "
-- Step 1: Define the PolarDB source table
CREATE TEMPORARY TABLE `db1`.`sbtest1` (
  `id`   BIGINT,
  `c1`   STRING,
  `c2`   STRING,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector'     = 'mysql',
  'hostname'      = '{mysql_host}',
  'port'          = '{mysql_port}',
  'username'      = '{mysql_user}',
  'password'      = '{mysql_password}',  -- Use a secrets manager in production.
  'database-name' = 'db1',
  'table-name'    = 't1',
  'server-id'     = '10000-11000'        -- Must be unique across all synchronization links.
);

-- Step 2: Define the PolarSearch target table
CREATE TEMPORARY TABLE `dest` (
  `id`    BIGINT,
  `max_c` STRING,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector' = 'opensearch',
  'hosts'     = '{search_host}:{search_port}',
  'index'     = 'dest',
  'username'  = '{search_user}',
  'password'  = '{search_password}'     -- Use a secrets manager in production.
);

-- Step 3: Define the transformation and write it to the target
INSERT INTO `dest`
SELECT
    `t1`.`id`,
    MAX(`t1`.`c1`)
FROM `db1`.`t1` AS `t1`
GROUP BY `t1`.`id`;
");

If you create multiple Flink SQL synchronization links, each must use a unique server-id range in the WITH clause to avoid conflicts.

Manage synchronization links

• View all links:

  CALL dbms_etl.show_sync_link();

• View a specific link by ID (replace <sync_id> with the ID returned when you created the link):

  CALL dbms_etl.show_sync_link_by_id('<sync_id>')\G

  Output fields:

  *************************** 1. row ***************************
          SYNC_ID: crb5rmv8rttsg
             NAME: crb5rmv8rttsg
           SYSTEM: search
  SYNC_DEFINITION: db1.t1 -> dest
    SOURCE_TABLES: db1.t1
      SINK_TABLES: dest
           STATUS: active   -- active = synchronizing normally
          MESSAGE:           -- error message, if any
       CREATED_AT: 2024-05-20 11:55:06
       UPDATED_AT: 2024-05-20 17:28:04
          OPTIONS: ...

Modify a synchronization link

Use the same "new index + new link" approach as for search views.

1. Create a new synchronization link to a new PolarSearch index.

2. Run CALL dbms_etl.show_sync_link_by_id('<sync_id>') and wait until the synchronization latency drops to 0–1 seconds.

3. Switch your business queries from the old index to the new one.

4. Drop the old synchronization link.

Drop a synchronization link

CALL dbms_etl.drop_sync_link('<sync_id>');

State transitions when dropping:

• `active`: Changes to dropping first, then to dropped after resources are cleaned up.

• `dropped`: All link information is permanently deleted.

• Other states: Dropping is not supported.