Imported from KWDB/KaiwuDB-Agent-Skills (skills/kwdb-data-migration/SKILL.md). Install upstream with npx skills add KWDB/KaiwuDB-Agent-Skills --skill kwdb-data-migration. Copyright stays with the author.

KWDB Data Migration Skill

IMPORTANT: How to Use This Skill

This is an AI Agent Skill, NOT a library for manual coding. Here's how it works:

Language Support

ALWAYS respond in the same language the user uses. This skill fully supports both Chinese and English users:

• If user writes in Chinese, respond in Chinese
• If user writes in English, respond in English
• When displaying technical terms (e.g., JDBC, DDL, TIMESERIES, RELATIONAL), keep them in their original form
• Example Chinese responses are available in user-interaction-scenarios.zh.md

Correct Usage Pattern (Natural Language)

User: Help me migrate MySQL database to KaiwuDB

AI Agent:
1. Reads this SKILL.md to understand the migration workflow
2. Guides user step-by-step to collect necessary parameters:
   - KDTS server address (default: http://127.0.0.1:8989)
   - MySQL connection info (host, port, user, password, database)
   - KaiwuDB connection info (host, port, user, password, database, engine type)
   - Migration mode (full/schema-only/data-only)
3. Calls the Python scripts in `scripts/` as backend tools
4. Reports progress and results to user in natural language

What AI Agent Does

The AI Agent acts as an intelligent assistant that:

• [STEP] Understands user's migration intent from natural language
• [COLLECT] Collects all necessary parameters through guided questions
• [EXECUTE] Executes migration operations by calling KDTS API
• [VALIDATE] Validates parameters and handles errors automatically
• [REPORT] Reports progress and results in user-friendly format
• [CONFIRM] Asks for confirmation at critical steps (DDL execution, task termination)

What User Needs to Do

User only needs to:

• [DESCRIBE] Describe migration intent in plain language (e.g., "Migrate MySQL to KaiwuDB")
• [PROVIDE] Provide connection information when asked
• [CONFIRM] Confirm critical operations when prompted

Example Conversation

User: Help me migrate MySQL sales database to KaiwuDB time series

AI Agent:
  [INFO] Migration intent recognized: MySQL -> KaiwuDB (TIMESERIES)
  
  Please provide the following information:
  1. KDTS server address (default: http://127.0.0.1:8989)
  2. MySQL connection info:
     - Host:
     - Port (default 3306):
     - Username:
     - Password:
  3. KaiwuDB connection info (engine: TIMESERIES):
     - Host (default 127.0.0.1):
     - Port (default 26257):
     - Username (default root):
     - Password:
  
  [WARNING] Security reminder: Please ensure you have backed up both source and target databases before migration!

User: KDTS at http://192.168.1.100:8989, MySQL: root/password abc at 192.168.1.50:3306, KaiwuDB at 127.0.0.1:26257

AI Agent:
  [OK] Parameters collected, starting migration process...
  
  Step 1: Testing database connections...
  [OK] MySQL connection successful
  [OK] KaiwuDB connection successful
  
  Step 2: Reading MySQL metadata...
  [INFO] Found 5 tables: orders, customers, products, categories, inventory
  
  Step 3: Configuring time-series tags (TIMESERIES target)...
  For table orders, select the time column and PRIMARY TAGS (1-4):
  - Time column: order_time
  - PRIMARY TAGS: customer_id, product_id
  [WARNING] Verify no NULL values in the primary tag columns
  
  Step 4: Previewing DDL (KaiwuDB time series tables)...
  [Shows DDL content with TAGS / PRIMARY TAGS]
  
  [WARNING] About to execute DDL to create tables. Continue? (yes/no)

Python Scripts Purpose

The Python scripts in scripts/ are backend tools for the AI Agent. They provide low-level functions that the Agent calls during migration. Users do NOT need to read or write these scripts directly.

Python Dependencies

The scripts require minimal dependencies:

All other modules use Python standard library only (typing, json, re, logging, etc.).

Overview

This skill provides automated heterogeneous database migration to KaiwuDB / KWDB through KDTS REST API. Unlike the old version that only provided manual GUI guidance, this skill directly calls the KDTS API to automate the entire migration workflow.

Migration Path: KDTS REST API

• Supports 14 source types: MySQL, Oracle, PostgreSQL, SQL Server, ClickHouse, TDengine 2.x/3.x, InfluxDB 1.x/2.x, OpenTSDB, MongoDB, FTP, HDFS, KaiwuDB (KaiwuDB as source = data migration only)
• Full automation: connection test, schema migration (DDL), data migration, progress tracking

KDTS Server Configuration

Before any migration operations, determine the KDTS Server connection. Configuration uses multi-layer priority (highest to lowest):

Configuration Methods

1. Environment Variables (Recommended for CI/CD)

# Option A: Full URL
export KDTS_BASE_URL="http://your-kdts-server.com:8989"

# Option B: Separate host and port
export KDTS_HOST="your-kdts-server.com"
export KDTS_PORT="8989"

# Optional additional settings
export KDTS_API_PREFIX="/kdts/api/v1"  # Default
export KDTS_TIMEOUT="300"               # Default seconds
export KDTS_CONNECT_TIMEOUT="5"         # Default seconds

2. Explicit Parameter

client = KDTSClient(base_url="http://your-kdts-server.com:8989")

3. Configuration File (kdts_config.json) Create kdts_config.json in your project directory:

{
  "base_url": "http://your-kdts-server.com:8989",
  "api_prefix": "/kdts/api/v1",
  "timeout": 300,
  "connect_timeout": 5
}

4. Default (Fallback)

Default: http://127.0.0.1:8989
API Prefix: /kdts/api/v1

Configuration Detection

Use get_environment_info() to check current configuration:

from scripts import get_environment_info
info = get_environment_info()
print(f"Config source: {info['config_source']}")
print(f"Current config: {info['current_config']}")

Mandatory Step

Ask the user for KDTS server address if:

• No environment variables are set
• No config file exists
• Default is not appropriate for their environment

Example prompt:

"What is your KDTS server address? (Default: http://127.0.0.1:8989)"

Script Reference

All migration operations use Python scripts in scripts/. Read scripts/README.md for API details.

Initialization

from scripts import (
    KDTSClient, DataSourceManager, MigrationWorkflowManager,
    get_environment_info
)

# Check current configuration
print(get_environment_info())

# Initialize client (uses multi-layer config: env > param > file > default)
client = KDTSClient()  # Reads from env or defaults to http://127.0.0.1:8989

# Or specify explicitly
client = KDTSClient(base_url="http://your-kdts-server:8989")

# Initialize managers
ds_manager = DataSourceManager(api_client=client)
workflow = MigrationWorkflowManager(api_client=client)

Config Methods

API Client Methods

Data Source Methods

Workflow Methods

Utility Methods

Mandatory Rules

1. Never Guess Parameters

All migration parameters must be collected from the user explicitly:

• KDTS server address (default: http://localhost:8989)
• Source database: engine, type, host, port, username, password, database name
• Target KWDB: engine, host, port, username, password, database name
• Migration scope: full database or specific tables
• Migration mode: schema-only, data-only, or full

2. Always Validate Source Type

Before any operation, must call ConfigValidator.validate_source_config() from scripts/config_validator.py:

• Check if source type is in supported list (14 types)
• Check if source type supports the requested operation (metadata, full migration, etc.)
• Refer to references/source-types.md for full capability matrix

3. Always Test Connection First

Before reading metadata or building migration scripts:

from scripts.api_client import KDTSClient
client = KDTSClient(base_url)

# Test source connection
result = client.test_connection(source_config, is_target=False)
if result['code'] != 0:
    raise Exception("Source connection failed")

# Test target connection  
result = client.test_connection(target_config, is_target=True)
if result['code'] != 0:
    raise Exception("Target connection failed")

NOTE (KDTS behavior): test_connection() returns code=0 even for FAILED validations — the failure text is in the data field. The api_client normalizes such responses to code=2001 automatically, so code == 0 always means success. Do NOT bypass this check.

If connection fails, stop immediately and show error hint from error_handler.py.

4. Mandatory Backup Reminder

Before any migration starts:

Reminder: Please ensure you have backed up both source and target databases before proceeding with migration. KDTS migration is non-transactional for data operations and cannot be automatically rolled back.

5. Never Kill Running Tasks Without Confirmation

CRITICAL: Never execute control_task(action="KILL") without explicit user confirmation:

1. Show current task status and progress
2. Warn: "Killing a running migration may leave data in inconsistent state"
3. Ask: "Are you absolutely sure you want to kill this task? (type 'YES' to confirm)"
4. Only proceed after explicit confirmation

6. Migration Task Naming Convention

When building scripts, inform user of the generated script names:

Script naming: <SOURCE>2<TARGET>_<timestamp>.json
Example: MYSQL2KAIWUDB_1719290000000.json

7. Engine Compatibility Rules (STRICT)

CRITICAL: Certain source types are STRICTLY limited to specific target engines. Do NOT attempt to bypass these restrictions.

Violation Handling: If user requests invalid combination (e.g., TDengine → RELATIONAL):

1. Explain the restriction clearly
2. Suggest alternative: Use native ETL tools or custom scripts for cross-engine migration
3. Show supported alternatives: Migrate TDengine → TIMESERIES, or export data manually then import to RELATIONAL

Supported Data Sources

Refer to references/source-types.md for complete capability matrix.

Note:

• Target is ALWAYS KaiwuDB with engine specified as RELATIONAL or TIMESERIES
• Source MUST also specify engine field (RELATIONAL for RDBMS, TIMESERIES for others)
• For SQL Server, InfluxDB 1.x/2.x: Use two-step migration (Schema first, then Data)
• Sources without metadata support (e.g. ClickHouse): KDTS preview_ddl generates DDL from the passed-in Database object, NOT from the source connection — so a table-based source (ClickHouse, TDengine 2.x) can still get DDL. REQUIRED interaction — the table structure MUST come from the USER (source CREATE TABLE DDL or a column list); NEVER guess the structure or rely on test code. Build the Database object manually from the user-provided structure (use build_manual_metadata()), then call preview_ddl() as usual. File sources (FTP/HDFS) have no table structure — pre-create the target tables.

KaiwuDB Time-Series Table Constraints

When migrating to KaiwuDB with TIMESERIES engine, the following constraints apply:

Primary Tag Type Rules (from KDTS source - TypeMapping.FLOAT_TYPE_NAMES):

• NOT eligible (float types): FLOAT, FLOAT4, FLOAT8, DOUBLE, REAL, BINARY_FLOAT, BINARY_DOUBLE, DECIMAL, NUMERIC → Auto-demoted to ordinary tags
  • Note: DECIMAL and NUMERIC are classified as float types by KDTS and cannot be primary tags
• NOT eligible: NULL/Nullable columns → Auto-demoted to ordinary tags
• Auto-converted: NVARCHAR, NCHAR, TEXT, CLOB, BLOB, BYTES, VARBYTES, JSON, ARRAY, MAP, INET, INTERVAL, UUID → Converted to VARCHAR(128)
• VARCHAR handling: Default 64 bytes, max 128 bytes (auto-truncated if exceeded)

Tag Type Rules (from KDTS source):

• Auto-converted: TIMESTAMP, TIMESTAMPTZ, NVARCHAR, GEOMETRY → Converted to VARCHAR

KDTS Auto-Mapping Algorithm:

1. Collect all source tags
2. Validate count <= 132
3. Demote invalid primary tags (FLOAT, DOUBLE, DECIMAL, NUMERIC, NULL)
4. Identify eligible primary tags (NOT NULL, NOT FLOAT/DECIMAL/NUMERIC, NOT over-length)
5. If 0 eligible → ERROR 3006
6. Select first N eligible as PRIMARY TAGS (max 4)
7. Auto-convert invalid types to supported types

Recommendation: When source has many columns, consider splitting into multiple tables or migrations.

Note: For complete DDL syntax and auto-mapping details, see references/ddl-syntax.md

API Endpoint Mapping

All endpoints under {base_url}/kdts/api/v1:

Migration Workflows

Workflow 1: Full Migration (Schema + Data)

When to use: Source supports full migration:

• MYSQL, ORACLE, POSTGRESQL, CLICKHOUSE, KAIWUDB, TDENGINE3X

Note:

• KAIWUDB and CLICKHOUSE support auto-discovery (Full Migration) but do NOT support metadata reading (DDL generation). You may need to pre-create target tables or use alternative DDL generation methods.
• SQLSERVER, INFLUXDB1X/2X support metadata + data but NOT auto-discovery (use Workflow 2).
• TDENGINE2X, OPENTSDB, MONGODB, FTP, HDFS only support data migration (use Workflow 3).

1. Collect parameters (interactive)
   +-- KDTS base URL
   +-- Source config (engine, type, host, port, user, password, db)
   |   Note: engine MUST be specified (RELATIONAL for RDBMS, TIMESERIES for others)
   +-- Target config (engine: RELATIONAL or TIMESERIES, host, port, user, password, db)
   |   Note: engine MUST be specified for KaiwuDB target
   +-- Metadata options (PK, constraint, comment, index, view)

2. Validate source type → ConfigValidator.validate_source_config()

3. Test connections → test_connection() × 2

4. Check target DB exists → list_databases()
   If not exists, remind user to create or use DDL

5. Read source metadata → read_metadata()
   Show table count, columns per table, PK/constraint info
   
   **Branch for sources WITHOUT metadata support**:
   5a. Check whether the TARGET table already exists (ask the user, or infer from a previous migration). 
   5b. If the target table exists and matches: skip DDL, go directly to data migration.
   5c. If NOT exists: **REQUIRED interaction — collect the table structure from the USER**
       (source CREATE TABLE DDL or a column list with names and types). NEVER guess the structure.
   5d. Build the Database object manually from the user-provided structure:
       ```python
       from scripts import build_manual_metadata, build_added_column, mark_time_series_columns
       db = build_manual_metadata('CLICKHOUSE', 'clickhouse_kwdb', 'test_tb', user_columns)
       mark_time_series_columns(db, 'test_tb', time_column='ts', primary_tags=['t1'], tags=[])
       # add a new column if the user wants one (e.g. t1 default 1)
       db['tableMap']['test_tb']['columns'].append(
           build_added_column('t1', 1, source_type='CLICKHOUSE', is_tag=True, is_primary_tag=True))
       ```
   5e. Continue with DDL preview (step 7) → confirmation → execute → then data migration.
       For file sources (FTP/HDFS, no table structure): pre-create the target tables.

6. **Configure Tags for Time-Series Target** (ONLY for TIMESERIES target with RELATIONAL source)
   
   **Trigger Condition**: Target engine is TIMESERIES AND source type is RELATIONAL (MySQL, Oracle, etc.)
   
   **Interaction Flow**:
   
   6.1 Show all columns for each table:
   ```
   Table: orders
   Columns:
   - id (BIGINT, PK)
   - customer_id (BIGINT)
   - product_id (BIGINT)
   - order_time (TIMESTAMP)
   - status (VARCHAR(50))
   - total_amount (DECIMAL(15,2))
   ```
   
   6.2 Ask user to select primary tags (1-4 columns, REQUIRED):
   ```
   Primary Tag Selection (1-4 required, max 4):
   [ ] id
   [ ] customer_id
   [ ] product_id
   [ ] order_time
   [ ] status
   [ ] total_amount
   
   Note: Primary tags are used for indexing and filtering in time-series queries
   Recommended: Select unique identifiers like device_id, sensor_id, etc.
   ```
   
   6.3 Ask user to select secondary tags (optional):
   ```
   Secondary Tag Selection (optional):
   [ ] id
   [ ] customer_id
   [ ] product_id
   [ ] order_time
   [ ] status
   [ ] total_amount
   
   Note: Secondary tags are additional indexed columns
   Recommended: Select commonly filtered columns like status, type, etc.
   ```
   
   6.4 Show summary and confirm:
   ```
   Tag Configuration Summary for orders table:
   - PRIMARY TAGS: customer_id, product_id
   - SECONDARY TAGS: status
   - VALUE FIELDS: id, order_time, total_amount
   
   Note: This configuration will be shown in the DDL preview.
   KDTS will generate appropriate time-series DDL based on this selection.
   ```
   
   **For Time-Series Sources (InfluxDB, TDengine, OpenTSDB)**:
   - Tags are AUTO-MAPPED from source to KaiwuDB:
     - InfluxDB tags → PRIMARY TAGS (first 4 eligible, rest become SECONDARY)
     - InfluxDB fields → VALUE columns
     - TDengine TAG columns → PRIMARY TAGS
     - TDengine regular columns → VALUE columns
   - No user interaction needed, but show the mapping in DDL preview for confirmation
   
   **Constraints**:
   - Maximum 4 PRIMARY TAGS per table (Error 3004 if exceeded)
   - At least 1 PRIMARY TAG required (Error 3006 if missing)
   - Maximum 4096 columns total (tags + values), max 132 tags from source
   - Column names max 128 bytes
   
   6.5 **Check Tag Column NULL Values (CRITICAL)**
   
   PRIMARY TAGS must be NOT NULL. If the source data contains NULL values in a column
   selected as PRIMARY TAG, the data migration will FAIL on write.
   
   **Required interaction**: After tag selection, check the source data for NULLs in
   all selected PRIMARY TAG columns:
   
   ```
   [WARNING] Primary Tag NULL Check
   ===============================
   Table: orders
   PRIMARY TAGS: customer_id, product_id
   
   Please verify in the source database (e.g., MySQL):
   SELECT COUNT(*) FROM orders WHERE customer_id IS NULL OR product_id IS NULL;
   
   If count > 0, options:
   1. Fix/backfill the NULL values in the source data, then re-run migration
   2. Choose different columns as PRIMARY TAGS (or demote to ordinary TAGS)
   3. Keep as-is — migration will fail on NULL tag values (NOT recommended)
   ```
   
   **Also apply to KDTS auto-mapped primary tags** (time-series sources): KDTS demotes
   NULL columns to ordinary tags with a warning, so the resulting DDL preview must be
   checked before execution.
   
   6.6 **Apply Tag Marks to the Metadata (CRITICAL)**
   
   KDTS generates time-series DDL from the tag marks on the source Database columns.
   After the user selects PRIMARY TAGS / TAGS / time column in steps 6.1-6.3, update the
   `source_db` object from `read_metadata()` — set these JSON fields per column
   (field names are declared explicitly in KDTS via `@JsonProperty`, matching
   `Column.java`; see `KaiwuDBStrategy.java` for the generation logic):
   
   - Time column: `"isTs": true` (KDTS renders it as the FIRST column, `TIMESTAMPTZ NOT NULL`)
   - Primary tag: `"isTag": true` AND `"isPrimaryTag": true` AND **`"nullAble": false`**
   - Ordinary tag: `"isTag": true` only
   - Everything else: leave `"isTs"/"isTag"/"isPrimaryTag"` as `false`
   
   **Primary tags MUST be NOT NULL in the column definition**: 
   KDTS demotes nullable primary tags to ordinary tags; if none remain eligible
   → Error 3006 (NO_PRIMARY_TAG). The helper sets `nullAble=false` automatically for
   primary tags — only use columns whose source DATA is NULL-free (see step 6.5).
   
   Use the provided helper to avoid manual field edits:
   ```python
   from scripts.api_client import mark_time_series_columns
   source_db = mark_time_series_columns(
       source_db=source_db,          # Database object from read_metadata()
       table_name="orders",
       time_column="order_time",
       primary_tags=["customer_id", "product_id"],
       tags=["status"],
   )
   ```
   
   **KDTS behavior when generating (from source code)**:
   - Tables with NO `"isTag": true` column are **SKIPPED** — no DDL is emitted for them
   - Primary tag demotion: FLOAT/DOUBLE/DECIMAL/NUMERIC or nullable columns → demoted to
     ordinary tags automatically (nullable primary tags are demoted with a warning)
   - If no eligible primary tag remains → Error 3006 (NO_PRIMARY_TAG)
   - Type conversion: primary tags of non-VARCHAR variable-length types (NVARCHAR, TEXT,
     CLOB, BLOB, VARBYTES, JSON, etc.) → converted to VARCHAR(128); VARCHAR > 128 truncated;
     VARCHAR without length → VARCHAR(64)
   - Ordinary tags of forbidden types (TIMESTAMP, TIMESTAMPTZ, NVARCHAR, GEOMETRY) → VARCHAR
   - > 132 tag columns → Error 3004; tag name > 128 bytes → Error 3005
   - `CREATE TS DATABASE` is emitted; Database `interval`/`retentions` fields become
     `PARTITION INTERVAL` / `RETENTIONS` clauses when present
   
   **Implementation Note**: 
   - KDTS `preview_ddl` fully supports time-series DDL generation from the tag marks
     described above — NO need for the skill to hand-craft DDL
   - The request field is `"isTimeSeries": true` (explicitly declared via
     `@JsonProperty("isTimeSeries")` in `PreviewDdlRequest.java`)
   - After DDL execution, use **Data-Only Migration** (Workflow 3) for data transfer
   - For TIMESERIES → TIMESERIES: KDTS handles automatic tag mapping internally

7. Preview DDL → preview_ddl()
   Show generated DDL for each table
   
   **For RELATIONAL → TIMESERIES (KDTS-Generated DDL)**:
   - Call `preview_ddl(target_config, source_db, metadata, is_time_series=True)` —
     the source_db object already carries the tag/primaryTag/ts marks from Step 6.6
   - KDTS generates `CREATE TS DATABASE` + time-series tables with TAGS / PRIMARY TAGS
   - Tables without any tag-marked column are SKIPPED in the output (warn the user)
   - Display with clear tag annotations (PRIMARY TAG, SECONDARY TAG, VALUE FIELD)
   - Point out any auto-demotion / type conversions in the DDL
     (FLOAT/nullable primary tags demoted, NVARCHAR→VARCHAR(128), etc.)
   - User confirms before execution
   
   Example DDL for RELATIONAL → TIMESERIES (as generated by KDTS):
   ```sql
   CREATE TABLE orders
   (
       order_time TIMESTAMPTZ NOT NULL,
       id BIGINT,
       total_amount DECIMAL(15,2)
   )
   TAGS
   (
       customer_id BIGINT NOT NULL,
       product_id BIGINT NOT NULL,
       status VARCHAR(50)
   )
   PRIMARY TAGS (customer_id, product_id);
   ```
   
   **Note**: For complete DDL syntax and KDTS auto-mapping details, see `references/ddl-syntax.md`
   - TIMESTAMPTZ is preferred over TIMESTAMP for timezone support
   - First column MUST be TIMESTAMPTZ NOT NULL
   - PRIMARY TAGS must be in the TAGS list and NOT NULL
   - Max 4 PRIMARY TAGS per table
   - KDTS auto-converts/demotes invalid tag types (see Section 3 in ddl-syntax.md)

   **For TIMESERIES → TIMESERIES (KDTS Auto-Generated)**:
   - Call `preview_ddl(target_config, source_db, metadata, is_time_series=True)`
     to get KDTS-generated DDL
   - KDTS auto-selects first 4 ELIGIBLE tags as PRIMARY TAGS (not simply first 4)
   - Ineligible tags (FLOAT, NULL, over-length) are auto-demoted to ordinary tags
   - Invalid types (NVARCHAR, TEXT, etc.) are auto-converted to VARCHAR
   - Show the mapped DDL for user confirmation with warnings about conversions
   
   Example DDL for TIMESERIES → TIMESERIES (InfluxDB auto-mapped):
   ```sql
   CREATE TABLE cpu_usage
   (
       time TIMESTAMPTZ NOT NULL,
       usage DOUBLE,
       temperature DOUBLE
   )
   TAGS
   (
       host VARCHAR(100) NOT NULL,
       region VARCHAR(50) NOT NULL
   )
   PRIMARY TAGS (host, region);
   ```
   
   Ask user to confirm before execution

8. Execute DDL → execute_ddl()
   **For RELATIONAL → TIMESERIES**: Execute the previewed DdlScript via the KDTS `execute_ddl()` API:
   `execute_ddl(target_config, previewed_ddl_script, auto_ddl=true)`.
   Do NOT rely on a direct JDBC/ODBC connection — the agent executes through KDTS.
   Note: the `createDb` field of DdlScript is NOT executed — KDTS generates the 
   CREATE DATABASE / CREATE TS DATABASE statement itself from the target engine and executes it when `auto_ddl=true`.
   **For TIMESERIES → TIMESERIES**: Use KDTS `execute_ddl()` API with the previewed DdlScript
   
   Report success with table count

9. **Switch to Data-Only Migration (RELATIONAL → TIMESERIES only)**
   Since the schema was created by the Skill-generated DDL, continue with **Workflow 3: Data-Only Migration**
   Note: `build_migration()` has NO `dataMode` parameter — data-only migration simply means
   providing **explicit table mappings** (`tables` REQUIRED, never empty) with the target tables already existing
   IMPORTANT: DataX configuration with `core` and `setting` is REQUIRED for successful data migration!
   
   Ask user: "Use default DataX configuration or customize?"
   
   Default DataX Configuration (Fixed Channel Count):
   ```json
   {
     "batchSize": 1000,
     "core": {
       "transport": {
         "channel": {
           "speed": {
             "byte": 1048576,
             "record": 1000
           }
         }
       }
     },
     "enable": true,
     "fetchSize": 1000,
     "setting": {
       "errorLimit": {
         "percentage": 0.02
       },
       "speed": {
         "channel": 4
       }
     }
   }
   ```
   Optional Configuration (Byte and Record Rate Limiting, Auto-Calculate Channel Count):
   ```json
   {
     "enable": true,
     "fetchSize": 1000,
     "batchSize": 1000,
     "core": {
       "transport": {
         "channel": {
           "speed": {
             "byte": 10485760,
             "record": 5000
           }
         }
       }
     },
     "setting": {
       "speed": {
         "byte": 52428800,
         "record": 40000
       },
       "errorLimit": {
         "percentage": 0.02
       }
     }
   }
   ```
   Note: The above configuration will auto-calculate channel count = min(52428800/10485760, 40000/5000) = min(5, 8) = 5
   
   If user wants to customize, explain the configuration based on KDTS source annotations:
   
   **UserData Top-Level Configuration**:
   
   | Field | Type | Default | Description |
   |---|---|---|---|
   | enable | boolean | false | Whether to enable user data migration |
   | fetchSize | int | 1000 | Number of records fetched per pull from source |
   | batchSize | int | 1000 | Number of records submitted per push to target |
   | core | Object | - | DataX core config (required) |
   | setting | Object | - | DataX setting config (required) |
   
   **core.transport.channel.speed Configuration** (Map<String, Object> - Per-Channel Level):
   `byte` and `record` can be configured simultaneously; they are different dimensions of rate limiting and **NOT mutually exclusive**!
   
   | Key | Type | Description |
   |---|---|---|
   | byte | Long | Per-channel byte rate limit (bytes/second), e.g., 1048576 means 1MB/s/channel |
   | record | Long | Per-channel record rate limit (records/second), e.g., 1000 means 1000 records/s/channel |
   
   **setting.speed Configuration** (Map<String, Object> - Global Level):
   The following parameters can be combined to implement flexible rate limiting strategies:
   
   | Key | Type | Description |
   |---|---|---|
   | channel | Integer | Fixed channel count. If configured, channel count is fixed and does not participate in auto-calculation |
   | byte | Long | Global byte rate limit, must be used with core.transport.channel.speed.byte |
   | record | Long | Global record rate limit, must be used with core.transport.channel.speed.record |
   
   **Configuration Combination Examples**:
   
   | Configuration Method | setting.speed Configuration | core.transport.channel.speed Configuration | Description |
   |---|---|---|---|
   | Fixed Channel Count + Global Rate Limit | channel=4, byte=52428800, record=40000 | byte=1048576, record=1000 | Fixed 4 channels, global rate limit distributed to each channel |
   | Byte-Only Rate Limit | byte=52428800 | byte=1048576 | Channel count auto-calculated = 52428800 ÷ 1048576 = 5 |
   | Record-Only Rate Limit | record=40000 | record=1000 | Channel count auto-calculated = 40000 ÷ 1000 = 40 |
   | Combined Byte and Record Rate Limit | byte=52428800, record=40000 | byte=1048576, record=1000 | Calculate channel count separately, take the larger value max(5, 40) = 40 |
   
   **Configuration Rules**:
   - If `setting.speed.byte` is configured, `core.transport.channel.speed.byte` **must** also be configured
   - If `setting.speed.record` is configured, `core.transport.channel.speed.record` **must** also be configured
   - channel only: Fixed channel count, per-channel rate limit controlled by core.transport.channel.speed
   - byte or record only: Auto-calculate channel count = global rate limit / per-channel rate limit
   - byte and record together: Calculate required channel count separately, take the larger value
   - channel and byte/record together: Channel count fixed, byte/record serve as global rate limits
   
   **setting.errorLimit Configuration** (Map<String, Object>):
   
   | Key | Type | Description |
   |---|---|---|
   | record | Integer | Maximum allowed number of error records |
   | percentage | Float | Maximum allowed error percentage, e.g., 0.02 means 2% |
   
   **Source-Specific Options (When Configuring Table Mapping)**:
   
   *RDBMS Sources (MySQL, Oracle, PostgreSQL, etc.):*
   - splitPk: Split primary key; when enabled, DataX uses concurrent fetching (primary key type must be numeric or string)
   - where: Filter condition, appended to SQL WHERE clause (mutually exclusive with querySql)
   - querySql: Custom query SQL array (mutually exclusive with where)
   
   *KaiwuDB Target:*
   - writeMode: Write mode, insert (default) or upsert
   - preSql/postSql: SQL executed before/after writing
   
   *Time-Series Sources:*
   - beginDateTime/endDateTime: Time range filter
   - splitIntervalS: Time window split (seconds) for concurrent fetching
   - tsColumn: Timestamp column name (required when using time range)
   
   **Configuration Rules**:
   - byte and record in `core.transport.channel.speed` can be configured simultaneously (different dimensions of rate limiting)
   - channel, byte, and record in `setting.speed` can be used in combination
   - If `setting.speed.byte` is configured, `core.transport.channel.speed.byte` must also be configured
   - If `setting.speed.record` is configured, `core.transport.channel.speed.record` must also be configured
   - where and querySql are mutually exclusive and cannot be used simultaneously
   - splitPk and querySql are not recommended to be used simultaneously (splitPk requires original table structure, querySql may have no table)
   
   **CRITICAL**: Do NOT skip this step! Missing core/setting config causes migration failures.
   
   For complete configuration reference, see `references/api-reference.md` Chapter 11.

10. Build migration script → build_migration_script(data_config=user_config)
    Show generated script name(s)
    **IMPORTANT — tables parameter by target engine**:
    - **RELATIONAL target**: `tables` can be empty (auto-discover all tables) for full migration
    - **TIMESERIES target**: `tables` MUST be explicit table mappings — empty tables fails with
      error 4001 "No datax contents generated from config" (KDTS source explicitly rejects 
      whole-database migration into the time-series engine)
    - **PostgreSQL source limitation**: auto-discovery filters tables by `schema == dbName`. 
      Tables in the `public` schema (the common case) are filtered out → auto-discovery 
      fails with 4001 even for RELATIONAL targets. Use explicit table mappings for PostgreSQL,
      unless the tables live in a schema named after the database.
    - **Oracle source requirements**:
      - **dbName must be the owner (schema) name, usually UPPERCASE** — KDTS reads Oracle
        metadata via `all_tab_comments WHERE owner = dbName`; a lowercase dbName returns
        zero tables (e.g. dbName=`ORACLE_KWDB`, not `oracle_kwdb`)
      - **Names are UPPERCASE**: table and column names from the metadata are uppercase
        (e.g. `TEST_TB`, `TS`, `C1`); table-mapping columns must match exactly (uppercase)
      - **New-column type trap**: when adding a column to the metadata, pick a
        sourceColumnType with an exact mapping — `NUMBER(5,0)→INT2`, `NUMBER(10,0)→INT4`,
        `NUMBER(19,0)→INT8`; an unmapped value like `NUMBER(1,0)` falls back to `NUMBER`
        → FLOAT (float type) → demoted primary tag → error 3006
      - **Expression columns need separated target columns**: source `"...,1 as t1"`
        requires `target_columns="...,t1"` (target must use real column names, else
        DataX cannot find the target column)
    - **Added-column type rules (ALL source types → KaiwuDB)**: when adding a
      column that the source lacks (e.g. a tag column), derive its type from the
      DEFAULT VALUE — use `build_added_column(column_name, default_value, source_type)`:
      - int default → INT4 (INT8 for InfluxDB), str default → VARCHAR, bool default →
        BOOL (eligible for PRIMARY TAG)
      - **float default → FLOAT4/FLOAT8 — ordinary TAG ONLY, NEVER a primary tag**
        (float types are demoted by KDTS; 3006 if no eligible primary tag remains)
      - applies to ALL sources: RDBMS (MySQL/Oracle/PostgreSQL/SQL Server/ClickHouse),
        TDengine 2.x/3.x, InfluxDB 1.x/2.x, KaiwuDB
      - the sourceColumnType is picked per source for an EXACT mapping
        (MySQL/SQLServer/TDengine `INT`, Oracle `NUMBER(10,0)`, PostgreSQL `INTEGER`,
        ClickHouse `INT32`, InfluxDB `INTEGER`, KaiwuDB `INT4`)
      - NOTE: KaiwuDB has no type-mapping rules in KDTS — verify the generated DDL
      - for SELECT-based sources the mapping source column uses a SQL expression
        matching the default (e.g. `1 as t1` for default 1); InfluxDB mappings use
        `build_influxdb_mapping()` (no SQL-expression support)
      Build mappings with `build_table_mapping()` per table (source + target columns).
      **Source identifier field per type**:
      `table` for RDBMS/KAIWUDB/TDENGINE/OPENTSDB, `measurement` for INFLUXDB,
      `collectionName` for MONGODB — build_table_mapping handles this automatically.
      FTP/HDFS are file sources (no table) — build their mappings manually (path-based).
      **InfluxDB REQUIRED interaction — data time range**: before building an
      InfluxDB mapping, ask the user for the data time range
      (begin/end "YYYY-MM-DD HH:MM:SS") and pass it to `build_influxdb_mapping()`
      with `split_interval_s` (default 86400 = 1 day). NO defaults for the range:
      a null range fails the migration, and a too-wide range (e.g. 1970~2099)
      causes reader memory overflow.
      Optional: readTimeout/connectTimeout (seconds; KDTS tests use 60).
      **Extended mapping options**: `build_table_mapping()`
      supports `where` (source filter, e.g. time range; RDBMS/KAIWUDB/TDENGINE/OPENTSDB),
      `pre_sql`/`post_sql` (target pre/post SQL, e.g. drop/create table), and SQL
      expression columns for RDBMS (e.g. `"...,1 as t1"`).
      **OpenTSDB mapping requirements**:
      - `column` is a list of FULL METRIC names in `table.metric` format
        (e.g. `"test_tb.c1,test_tb.c2,..."`) — NOT plain column names
      - **time range REQUIRED**: beginDateTime/endDateTime (no splitIntervalS) —
        collect the data time range from the user
      - OpenTSDB usually has NO authentication (username/password may be empty)
      **SQL Server source requirements**:
      - JDBC URL MUST include `encrypt=true;trustServerCertificate=true`
        (modern JDBC drivers require TLS by default):
        `jdbc:sqlserver://host:1433;databaseName=db;encrypt=true;trustServerCertificate=true`
      - supports `where` filters and SQL expression columns (RDBMS mapping, e.g.
        `1 as t1`); two-step migration (schema + data), no full migration
      - **schemaName fix**: the metadata table schemaName may come back as
        the DATABASE name → DDL becomes `"db"."db"."table"` (duplicated). Set the table
        schemaName to `public` so DDL is `"db"."public"."table"`
      **KaiwuDB-source mapping requirements** (KaiwuDB→KaiwuDB): 
      the KaiwuDB source (time-series engine) REQUIRES `beginDateTime`/`endDateTime` (data time range)
      and `tsColumn` (time column name, e.g. "ts") in the mapping — collect the time range from the user.
      **MongoDB mapping requirements**:
      - source identifier is `collectionName` (NOT `table`)
      - `column` is a JSON array string with name/type (e.g.
        `[{"name":"ts","type":"date"},{"name":"c1","type":"int"},...]`), NOT a comma
        string — types: date/int/long/double/bool/string/bytes
      - **`query` (optional filter)**: MongoDB JSON query syntax as a string,
        e.g. `{"t1":{"$gte":1,"$lt":8}}` — filters documents before migration
      - **Table creation is LIMITED to two options** (KDTS does NOT support MongoDB→KaiwuDB 
        type mapping — its generic fallback emits invalid types like LONG/STRING):
        1. **User pre-creates the target table** (skip DDL, migrate data directly);
        2. **SKILL generates the DDL from the USER-provided column info + type
           mapping rules** (int→INT4, long→INT8, double→FLOAT8, string→VARCHAR,
           bytes→VARBYTES, date→TIMESTAMP, bool→BOOL), user confirms, then
           `execute_ddl()`. If DDL execution FAILS: inform the user that the target
           table was NOT created and END the migration.
      **FTP mapping requirements**:
      - **`path` MUST be an absolute path starting with `/`**
      - **`path` is the SFTP SERVER-side path**, NOT the client local path — it must
        be resolvable where the SFTP service runs (Windows host, WSL, or container).
      - **`skipHeader: true`** when the CSV has a header row (otherwise the header is migrated as data)
      - `column` is a JSON array string with index/type/format per field (see `references/config-templates.md` §10.1)
      **HDFS mapping requirements**:
      - `path` is the HDFS SERVER-side path (JSON array string, absolute, e.g.
        `["/user/hive/warehouse/hdfs_test.db/test_tb"]`) — NOT a local path
      - `fileType` REQUIRED: `text` / `orc` / `parquet` / `rcfile` (text supports
        fieldDelimiter/encoding/compress/csvReaderConfig; orc/parquet do not)
      - `column` is a JSON array string with index/type/format per field (same
        structure as FTP, e.g. 14 columns with date/long/double/boolean/string)
      - `compress`: file compression (e.g. `gzip` for text files)
      - Kerberos-secured clusters: `haveKerberos` + `kerberosPrincipal` + `kerberosKeytabFilePath`
      - DataSource: host = NameNode host, port = 9000 (RPC), user/password
      - target tables must pre-exist (file source, no table structure)
      **HDFS mapping example**: see `references/config-templates.md` §10.2
      **View migration**: to migrate views, pass metadata option `view: true` to
      `read_metadata()` / `preview_ddl()` — views are then included in the DDL.

11. Execute migration → execute_migration() or execute_migration_batches()
    Return log file paths
    **BATCH EXECUTION (REQUIRED for many scripts)**: When `build_migration()` returns
    MANY scripts (one per table), DO NOT submit them all in a single
    `execute_migration()` call — the KDTS server starts DataX processes sequentially
    and the request exceeds the client read timeout.
    Instead use the batch workflow manager:
    ```python
    result = workflow.execute_migration_batches(
        script_names=script_names,   # all generated scripts
        batch_size=10,               # 10 scripts per batch
        batch_timeout=3600,          # per-batch wait limit
    )
    # result['all_succeeded'] == True  → every batch reached SUCCEEDED
    ```
    Rule of thumb: batch_size=10 for large table counts (>20 tables); single-shot
    `execute_migration()` is fine for ≤ 10 scripts. Timeout note: a 4003 on submission
    means the request reached the server (it keeps processing) — still monitor.

12. Monitor progress → query_task_status() (polling every 2s)
    Show status: SUBMITTED → RUNNING → SUCCEEDED/FAILED
    Report final status (execute_migration_batches monitors each batch to completion)

13. Verify (manual step for user)
    Remind to compare row counts between source and target

Workflow 2: Schema-Only Migration

When to use: Only need table structure, no data transfer

Steps 1-8 from Workflow 1, then STOP.
Report DDL execution result.

Workflow 3: Data-Only Migration

When to use:

• Target tables already exist, only need data sync
• For InfluxDB 1.x/2.x: Use this after Schema migration (Workflow 1 steps 1-7)

1. Collect parameters (interactive)
   +-- KDTS base URL
   +-- Source config
   +-- Target config
   +-- Table mappings (tables MUST be provided)

2. Validate source type

3. Test connections × 2

4. **Configure DataX Parameters** (REQUIRED - see Workflow 1 step 8 for details)
   Show default config and ask user to confirm or customize

5. Build migration script with explicit tables field
   → build_migration(tables=table_mappings, data_config=user_config)
   Note: tables are ALWAYS required for TIMESERIES targets (auto-discovery fails with 4001)

6. Execute migration (use `execute_migration_batches()` when > 10 scripts — see
   Workflow 1 step 11 for batch execution rules)

7. Monitor progress

Workflow 4: Table-Level Migration (Restricted Sources)

When to use: Source does NOT support full migration (SQL Server, TDengine 2.x, OpenTSDB, MongoDB, FTP, HDFS)

1. Collect ALL table mappings explicitly:
   Source: table name, columns
   Target: table name, columns, write mode (insert/upsert)

2. Validate source type

3. Test connections × 2

4. **Configure DataX Parameters** (REQUIRED - see Workflow 1 step 8 for details)
   Show default config and ask user to confirm or customize

5. Build migration script with explicit tables field
   → build_migration(tables=all_mappings, data_config=user_config)

6. Execute and monitor (use `execute_migration_batches()` when > 10 scripts — see
   Workflow 1 step 11 for batch execution rules)

Source Type Configuration Templates

Source Configuration Examples

Important: For ALL source configurations, the engine field is REQUIRED per KDTS API:

• Use RELATIONAL for: MySQL, Oracle, PostgreSQL, SQL Server, ClickHouse
• Use TIMESERIES for: KAIWUDB, TDengine 2.x/3.x, InfluxDB 1.x/2.x, OpenTSDB, MongoDB, FTP, HDFS

Relational Source (MySQL Example)

{
  "engine": "RELATIONAL",
  "type": "MYSQL",
  "host": "127.0.0.1",
  "port": 3306,
  "username": "root",
  "password": "********",
  "dbName": "source_db"
}

Time Series Source (InfluxDB Example)

{
  "engine": "TIMESERIES",
  "type": "INFLUXDB1X",
  "host": "127.0.0.1",
  "port": 8086,
  "username": "admin",
  "password": "********",
  "dbName": "source_db"
}

Target Configuration Examples

Note: For target (KaiwuDB) configuration, engine field is REQUIRED to specify the KaiwuDB storage engine:

• Use RELATIONAL for relational database
• Use TIMESERIES for time-series database

KaiwuDB Target - Relational Engine

{
  "type": "KAIWUDB",
  "engine": "RELATIONAL",
  "host": "127.0.0.1",
  "port": 26257,
  "username": "root",
  "password": "********",
  "dbName": "target_db",
  "isTarget": true
}

KaiwuDB Target - Time Series Engine

{
  "type": "KAIWUDB",
  "engine": "TIMESERIES",
  "host": "127.0.0.1",
  "port": 26257,
  "username": "root",
  "password": "********",
  "dbName": "target_ts_db",
  "isTarget": true
}

Source Type → sourceType Mapping

When building migration scripts, use the appropriate sourceType:

Error Handling

Refer to references/error-codes.md for complete error code reference.

When API returns error:

1. Extract code and message from response
2. Call get_error_hint(code) to get user-friendly explanation and fix suggestion
3. Show both original error and hint to user
4. If it's a connection/validation error, stop and ask for corrected parameters
5. If it's a data migration error, show partial progress and ask whether to retry or skip

Common Error Scenarios

Interactive Parameter Collection

When user intent is identified but parameters are missing, collect them step by step:

Step 1: KDTS Server

What is the KDTS server address?
(default: http://localhost:8989)

Step 2: Migration Type

Select migration type:
1. Full Migration (schema + data)
2. Schema-Only Migration (DDL only)
3. Data-Only Migration (tables must exist)

Step 3: Source Configuration

Source database type?
[MySQL, Oracle, PostgreSQL, SQL Server, ClickHouse, TDengine 2.x/3.x, 
 InfluxDB 1.x/2.x, OpenTSDB, MongoDB, FTP, HDFS, KaiwuDB]

Source connection:
Host: 
Port: (show default based on type, e.g., MySQL=3306)
Username: 
Password: 
Database:

Step 4: Target Configuration

Target KaiwuDB connection:
Engine: [RELATIONAL, TIMESERIES]
Host: (default: 127.0.0.1)
Port: (default: 26257)
Username: (default: root)
Password: 
Database:

Step 5: Migration Scope

Migration scope:
1. Full database (all tables)
2. Specific tables only

If specific tables:
- Table name(s):
- Columns (optional):

Step 6: Data Configuration (Optional)

Data migration settings:
Fetch size (rows per fetch, default 1000): 
Batch size (rows per write, default 1000):
Error tolerance (% allowed, default 0.02):
Concurrency (channels, default 4):

Confirmation Gates (Critical Safety Points)

Gate 1: DDL Execution Confirmation

Before executing DDL, you MUST:

1. Show the previewed DDL to the user
2. Explain what will be created (databases, tables, indexes, constraints)
3. Warn about potential issues:
   • Existing tables will be overwritten (i

Truncated - read the full file at https://github.com/KWDB/KaiwuDB-Agent-Skills/blob/49e6e97bedf0c8a645a6731a6edb1c86fdeb4991/skills/kwdb-data-migration/SKILL.md.