KWDB Data Migration

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: 2_.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 - 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 - 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):
   
   | 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)
   - query