dolphindb-data-import

name	dolphindb-data-import
description	DolphinDB data import guide for IoT and general scenarios. Covers CSV/TXT, JSON, binary, plugin-based (HDF5, Parquet, MySQL, ODBC), and IoT real-time ingestion (MQTT, OPC UA). Includes IOTDB engine, date/time conversion, array vector import, null handling, and binary record import. Does NOT cover database administration, permission management, stream compute engine configuration, or data export.
version	1.1.0
category	dolphindb
argument-hint	[source type] [target database] [file path or connection] [IoT or general]
disable-model-invocation	false
user-invocable	true

DolphinDB Data Import Guide

Use this skill when the user asks about importing, loading, ingesting, or migrating data into DolphinDB, including CSV/text files, JSON, binary files, external databases, Parquet/HDF5 files, or IoT real-time data streams.

Import Method Decision Tree

1. CSV / text file → memory table: use loadText (single-thread) or ploadText (parallel, faster for files > 16 MB).
2. CSV / text file → distributed table: use loadTextEx (recommended, direct-to-disk, avoids OOM for large files).
3. Very large CSV file (> memory): use textChunkDS + mr to chunk and import.
4. JSON string: use fromStdJson (standard JSON) or parseJsonTable (JSON to table).
5. Binary file with string columns: use loadRecord(filename, schema). Schema must be provided by user.
6. Binary file without string columns: use readRecord!(handle, holder). Preferred for pure numeric files — simpler and safer.
7. Parquet file → memory table: use parquet::loadParquet(filePath, [schema], [columnsToLoad], [startRowGroup], [rowGroupNum]). Supports partial column loading via columnsToLoad and row group selection via startRowGroup/rowGroupNum.
8. Parquet file → distributed table: use parquet::loadParquetEx(dbHandle, tableName, partitionColumns, fileName, [schema], [columnsToLoad], [startRowGroup], [rowGroupNum], [transform]). Note: no sortColumns parameter; if sortColumns is needed, use parquet::loadParquet + replaceColumn! + append!.
9. HDF5 file → memory table: use hdf5::loadHDF5(fileName, datasetName, [schema], [startRow], [rowNum]). Supports row range selection via startRow and rowNum.
10. HDF5 file → distributed table: use hdf5::loadHDF5Ex(dbHandle, tableName, [partitionColumns], fileName, datasetName, [schema], [startRow], [rowNum], [transform]). Note: no sortColumns parameter; if sortColumns is needed, use hdf5::loadHDF5 + replaceColumn! + pt.append!(data).
11. External database (MySQL, ODBC, etc.): use the corresponding plugin.
12. IoT real-time data (MQTT): use MQTT plugin to subscribe to broker topics and write to stream tables.
13. IoT real-time data (OPC UA): use OPC UA plugin to subscribe to server nodes and write to stream tables.
14. IoT point management: use IOTDB engine with IOTANY column and latestKeyCache for millisecond-level latest-value queries.

Quick Reference: Core Import Functions

Function	Target	Key Feature
loadText	Memory table	Single-thread; schema, skipRows, delimiter, partial columns via schema.col
ploadText	Memory table	Parallel; faster for > 16 MB
loadTextEx	Database table	Direct-to-disk; transform, sortColumns, skipRows
textChunkDS	Chunks for mr	Splits large file; use with mr
extractTextSchema	Schema preview	Auto-detect column types and delimiter
loadRecord	Memory table	Binary import with string support; schema required
readRecord!	Memory table	Binary import, no string support; holder required
parquet::loadParquet	Memory table	columnsToLoad, startRowGroup, rowGroupNum
parquet::loadParquetEx	Database table	Direct-to-disk; columnsToLoad, startRowGroup, rowGroupNum, transform (no sortColumns)
hdf5::loadHDF5	Memory table	Schema override; startRow, rowNum
hdf5::loadHDF5Ex	Database table	Direct-to-disk; startRow, rowNum, transform (no sortColumns)

Critical Rules

• loadText/loadTextEx do NOT support rowNum parameter (⚠️ CRITICAL — #1 cause of wasted React rounds): loadText and loadTextEx do NOT have a rowNum parameter. To preview data, use select top N * from loadText(...) or load first then select top N. NEVER attempt to pass rowNum to these functions — it will cause an error and waste a React round.
• Delimiter auto-detection: When extractTextSchema returns all columns merged into a single STRING column, it means the delimiter was not correctly detected. Automatically try these delimiters in order: (1) Tab: char(9) (most common for TXT files), (2) Semicolon: ;, (3) Pipe: |. Re-run extractTextSchema(filename, delimiter=char(9)) etc. until columns are properly separated.
• skipRows semantics: skipRows=N skips the first N lines of the file, INCLUDING the header row. When using skipRows > 0, you MUST provide a schema parameter (from extractTextSchema) because the header row will be skipped. If the user wants to skip N data rows but keep the header, do NOT use skipRows — instead, load all data with loadText, then use select * from t where rowNo(t) >= N or filter after import. skipRows does NOT count the header row separately — it counts from line 1 of the file.
• Array vector import: When a CSV column contains delimited values (e.g., "1.0,2.0,3.0"), it should be imported as an array vector. Steps: (1) extractTextSchema → (2) modify schema type to DOUBLE[] (or INT[] etc.) → (3) use loadText(path, schema=schema, arrayDelimiter=","). The arrayDelimiter parameter tells DolphinDB how to split the values within each cell.
• temporalParse with insufficient digit count: When converting INT time values (e.g., 93000000) to TIME/TIMESTAMP using temporalParse, the string representation must match the format pattern length exactly. If the INT value has fewer digits than the format pattern requires, use lpad(string(col), N, "0") to pad with leading zeros. lpad is safe: it only pads strings shorter than N, and leaves strings already at length N or longer unchanged. Example: INT 93000000 (8 digits) → lpad(string(93000000), 9, "0") → "093000000" → temporalParse("093000000", "HHmmssSSS") → 09:30:00.000. INT 100000000 (9 digits) → lpad(string(100000000), 9, "0") → "100000000" (unchanged) → temporalParse("100000000", "HHmmssSSS") → 10:00:00.000. Always check digit count before applying temporalParse.
• File path MUST be provided first: If the user's message does NOT contain a file path (no string matching patterns like /path/, .csv, .txt, .parquet, .h5, .bin), you MUST use ask_user to ask for the file path BEFORE doing anything else. Do NOT attempt to execute any DolphinDB scripts until you have a file path. This is the FIRST step of the workflow and cannot be skipped.
• Always preview schema first: use extractTextSchema(filename) to check auto-detected types before importing. If types are wrong, modify the schema and pass it to the import function.
• Date literals in DolphinDB: use 2024.01.01 format, NOT '2024-01-01'.
• Long integer timestamps: import as LONG first, then convert with timestamp() function. Do NOT specify LONG timestamps as TIMESTAMP type directly in schema — it will produce NULL values.
• Timezone handling: use localtime() or convertTZ() to convert UTC timestamps to local time after import.
• Header with numbers: if column names start with digits, set containHeader=true to preserve names (system prefixes "c").
• Partition conflict: avoid parallel writes to the same partition. Use serial import or set atomic='CHUNK'.
• Partition size: recommended 100 MB to 1 GB per partition (uncompressed).
• Null value handling: use nullFill/nullFill! (fill with specific value), ffill (forward fill), bfill/bfill! (backward fill), interpolate (linear/pad/nearest/krogh interpolation), or lfill! (linear fill between endpoints). In loadTextEx, use transform parameter for one-step null filling during import.
• FLOAT and DOUBLE cannot be used as partition columns; convert STRING to SYMBOL for partitioning.
• Script blocks are not function libraries: copy script blocks from examples and replace placeholder variables (file paths, database names, table names); do not run entire example files, do not define or call custom functions.
• Interactive decision-making: when creating a distributed table, use ask_user to confirm storage engine and partition scheme before proceeding. Provide options with a recommended default. Do NOT ask about things you can determine yourself (e.g., file format from extension, schema from extractTextSchema, import method from file size).
• Partial column loading: For text files, use schema's col column to specify column indices (must be ascending). For Parquet, use columnsToLoad parameter (integer vector of zero-based column indices). For HDF5, load all columns then select.
• Row skipping: For text files, use skipRows (0-1024). For binary files, use skipBytes in loadRecord or offset/length in readRecord!. For Parquet, use startRowGroup and rowGroupNum to read specific row groups. For HDF5, use startRow and rowNum parameters in loadHDF5/loadHDF5Ex.
• Binary file schema is mandatory: DolphinDB cannot auto-detect binary file structure. ALWAYS obtain schema from user before using loadRecord or readRecord!. If user provides the writeRecord script, derive schema from it.
• Data import workflow: Follow the 10-step workflow (file path → format → metadata → user confirm → import method → preprocessing → final form → execute → verify → summary). Skip steps only when information is already provided or can be reliably inferred.
• Schema confirmation is MANDATORY (Step 4): NEVER skip schema confirmation, even when all types appear correct and no conversion is needed. The user may want to adjust types, and their confirmation takes priority.
• Memory table variable name confirmation: When importing to a memory table, you MUST ask the user to confirm the variable name (e.g., t in t = loadText(...)). Suggest a default name based on the file name (e.g., trades for trades.csv). Before assigning, check if the variable name already exists using try { objByName("varName", false) } catch(ex) { ... }. If the name exists: (i) Ask the user whether to overwrite (use undef(\varName)to delete the old variable, then reassign) or rename (provide a new name and validate uniqueness again). (ii) NEVER overwrite a variable without user confirmation. **⚠️ API note**:existsShareVariableonly checks shared variables — for local variables, use thetry { objByName(...) } catch(ex)pattern. To delete a local variable, useundef(`varName)`.
• Existing database/table handling (⚠️ CRITICAL): Before creating a distributed table, check if the database and table already exist. Use existsDatabase("dfs://path") to check database, then existsTable("dfs://path", "tableName") to check table. ⚠️ API pitfall: existsTable and listTables take a path string (e.g., "dfs://sensor_db"), NOT a database() object. db.listTables() will cause "db isn't an instance of a class" error — always use listTables("dfs://path") instead. If the database exists: (i) If existsTable returns false (table name differs from any existing table), import directly into the existing database without dropping it. (ii) If existsTable returns true (table name matches an existing table), determine the import mode by analyzing the prompt content: Append import mode (prompt explicitly mentions "append"/"追加", OR the table existed before this test flow — e.g., user pre-created the table for historical data accumulation): verify schema consistency between source data and existing table, adjust source data structure to match existing table schema, then append data. If schema mismatch or import fails, create a new table (preferably in the same database) and import there. New table import mode (prompt does NOT mention append, OR the table was created by a previous executor attempt in the same test flow): use truncate(dbPath, tableName) to clear existing data before import to avoid duplication from repeated executor attempts. In interactive mode, ask the user whether to append or overwrite. NEVER drop an existing database or table without explicit user confirmation.
• Non-interactive mode (⚠️ CRITICAL — for automated testing/batch processing): When the user's initial prompt provides ALL necessary information (file path, target form, database path, table name, engine, partition scheme, schema, type conversions, etc.), SKIP all ask_user confirmation steps and execute directly. If the target database already exists and the table name conflicts, use a new database path (e.g., append a suffix) instead of dropping the old database — this avoids irreversible operations. ⚠️ Testing platform restriction: The DolphinX testing platform PROHIBITS dropDatabase and dropTable operations. NEVER use these functions in testing scenarios — always use existsDatabase + existsTable conditional creation logic, or use a unique database path to avoid conflicts. ⚠️ Import mode handling when table exists: When the table already exists, determine the import mode by analyzing the prompt: (i) If the prompt explicitly mentions "append"/"追加" → append import mode: verify schema consistency between source data and existing table (column count, partition column inclusion, column type compatibility), adjust source data structure to match existing table schema, then append data. If schema mismatch or import fails, create a new table (preferably in the same database with a new name) and import there. (ii) If the prompt does NOT mention append → new table import mode: use truncate(dbPath, tableName) to clear existing data before import to avoid duplication from repeated executor attempts within the same test flow. truncate preserves the table structure and only clears data — it is NOT a drop operation and is safe to use on the testing platform. If truncate fails, append to existing data and report a warning.
• DB/Table existence 4-branch handling (⚠️ CRITICAL): When using loadTextEx + transform (or any import function that requires a pre-created table), MUST handle all 4 combinations of database/table existence:

  dbExists = existsDatabase(dbPath)
  tableExists = iif(dbExists, existsTable(dbPath, tableName), false)
  if (!dbExists) {
      // Branch 1: DB doesn't exist — create new DB and table
      db = database(dbPath, VALUE, ..., engine="OLAP")
      db.createPartitionedTable(schemaTable, tableName, partitionColumns)
  } else if (dbExists && !tableExists) {
      // Branch 2: DB exists but table doesn't — create table in existing DB
      db = database(dbPath)
      db.createPartitionedTable(schemaTable, tableName, partitionColumns)
  } else {
      // Branch 3 & 4: DB and table both exist
      db = database(dbPath)
      // Determine import mode by analyzing prompt:
      // - Append import (table existed before this test flow): verify schema, then append
      // - New table import (table from previous executor attempt): truncate to avoid duplication
      // Non-interactive mode defaults:
      // - If prompt mentions "append"/"追加": append mode (verify schema, then append)
      // - Otherwise: truncate mode (clear data, then import fresh)
      try { truncate(dbPath, tableName) } catch(ex) { /* truncate failed, append to existing data */ }
  }
  loadTextEx(db, tableName, partitionColumns, filePath, schema=schema, transform=transformFunc)
  

  Without this 4-branch logic, repeated runs will fail with "transform requires pre-created table" or "database already exists" errors. ⚠️ Data duplication prevention: In non-interactive mode, if the table already exists and the prompt does NOT explicitly request append import, use truncate(dbPath, tableName) to clear existing data before import. This prevents data duplication caused by repeated executor attempts within the same test flow. truncate preserves the table structure and only clears data — it is NOT a drop operation and is safe on the testing platform. If truncate fails (e.g., permission error), append to existing data and report a warning.
• transform function mutable rules (⚠️ CRITICAL): The transform function parameter is a constant reference by default.
  • Using update t set col = ... syntax: NO mutable keyword needed (recommended, simpler)
  • Using replaceColumn!(t, ...) or other mutable functions: parameter MUST be declared as mutable t
  • Example 1 (update syntax): def cleanTransform(t) { update t set volume = nullFill(volume, 0); return t }
  • Example 2 (mutable syntax): def convertMarketData(mutable t) { replaceColumn!(t, ts, timestamp(exec ts from t)); return t }`
  • WRONG: def convert(t) { replaceColumn!(t, col, ...) }` — will fail with "Constant variable [t] can't be used as argument for mutable function replaceColumn!"
• transform requires pre-created table (⚠️ CRITICAL): When using transform parameter with loadTextEx/loadParquetEx/loadHDF5Ex, the partitioned table MUST be created BEFORE calling the import function using db.createPartitionedTable(schemaTable, tableName, partitionColumns, [sortColumns]). loadTextEx with transform does NOT auto-create the table. Without transform, loadTextEx auto-creates the table.
• TSDB engine requires sortColumns (⚠️ CRITICAL): When using engine="TSDB", sortColumns parameter is MANDATORY in createPartitionedTable and loadTextEx. If sortColumns is not needed, use engine="OLAP" instead. OLAP engine does NOT support array vector types (DOUBLE[], INT[]) — for array vectors, MUST use TSDB with sortColumns.
• Engine-aware import verification (⚠️ CRITICAL — row count equality check fails for PKEY / TSDB+LAST/FIRST): After import, NEVER verify data correctness by simply comparing afterCount - beforeCount == expectedImportCount for ALL engines — this equality check is ONLY valid for non-overwrite scenarios (OLAP engine, or TSDB with keepDuplicates=ALL). For PKEY engine or TSDB with keepDuplicates=LAST/keepDuplicates=FIRST, the engine OVERWRITES existing rows with the same primary key (PKEY Merge-on-Write) or DEDUPLICATES rows with the same sortColumns value (TSDB+LAST/FIRST keeps only the last/first occurrence), so afterCount - beforeCount may be LESS than expectedImportCount even though the import succeeded. For these overwrite/deduplication scenarios, MUST use the three-step key-existence check (Strategy B) detailed in Step 9 of the workflow: (B1) afterCount >= beforeCount; (B2) afterCount - beforeCount <= expectedImportCount; (B3) extract unique key combinations via select distinct keyCols from sourceData, extract the same key combinations from the target table, use ej(uniqueKeys, targetKeys, keyCols)to find matched keys, and verifycount(matchedKeys) == count(uniqueKeys). **⚠️ NEVER rely on tableInsert/append!/loadTextExreturn values for verification**: For PKEY engines and key-value tables,tableInsertreturn value only counts NEW rows inserted (NOT rows updated due to key conflicts);append!andloadTextExdo not return inserted count at all. Always use SQLselect count(*)` and key-existence checks instead.
• Key columns for engine-aware verification: Determine the key columns for Strategy B based on the storage engine used at table creation time. (i) PKEY engine: use the columns declared in the primaryKey parameter of createPartitionedTable. (ii) TSDB engine with keepDuplicates=LAST/FIRST: use the columns declared in the sortColumns parameter. If sortColumns contains only ONE column that coincides with the partition column, append the partition column to form a composite key. (iii) For OLAP engine or TSDB with keepDuplicates=ALL, Strategy A (row count difference equality) is sufficient — no key-existence check needed.
• sortColumns format (⚠️ CRITICAL): sortColumns accepts column name vector, NOT comma-separated string.
  • Single column: sortColumns=date`` or sortColumns="date" (string scalar)
  • Multiple columns: sortColumns=date`sym`` (backtick vector, recommended)
  • Variable reference: sortColumns=colName (variable value is string or symbol)
  • WRONG: sortColumns="date,sym" (interpreted as single column name "date,sym")
  • WRONG: sortColumns=["date","sym"] (string array, should use backticks)
• loadHDF5Ex partition column type CANNOT be transformed (⚠️ CRITICAL): loadHDF5Ex uses the HDF5 file's schema to create the partitioned table, so the partition column type is FIXED to the HDF5 type (e.g., INT). The transform function only transforms data AFTER it's loaded, but the partition scheme is determined at table creation time.
  • If HDF5 has INT tradingDay but database requires DATE partition: loadHDF5Ex will ALWAYS fail with "partitioning column type doesn't match"
  • Solution: MUST use hdf5::loadHDF5 + replaceColumn! + createPartitionedTable (with correct DATE type) + pt.append!
  • Same rule applies to parquet::loadParquetEx
• IOTDB create table syntax (⚠️ CRITICAL — #1 cause of IOTDB import failure): IOTANY column can ONLY be created via SQL create table statement, NOT via table() function or createPartitionedTable.
  • Full syntax (SQL statement executed directly in DolphinDB script):

    create table `dfs://iot_db`.`sensors` (
        deviceId INT,
        location SYMBOL,
        timestamp TIMESTAMP,
        value IOTANY
    )
    partitioned by deviceId, timestamp
    sortColumns=`deviceId`location`timestamp
    latestKeyCache=true
    

  • Identifiers MUST use backticks (`), NOT double quotes (")
  • partitioned by is followed by comma-separated column names (NOT backtick vector)
  • sortColumns uses backtick vector
  • latestKeyCache=true is a key-value pair connected with =
  • Common error: schema must be provided to create table indicates syntax parsing failure — check backticks and keywords
• HDF5 plugin: ls vs lsTable (⚠️ CRITICAL): hdf5::ls and hdf5::lsTable return DIFFERENT column structures:
  • hdf5::ls(path) returns table with columns [objName, objType] — for exploring file structure
  • hdf5::lsTable(path) returns table with column [tableName] — for listing datasets
  • DO NOT mix column names: ls has no name or tableName column; lsTable has no objName or objType column
  • Recommended: use hdf5::lsTable(path) to get dataset names, then use tableName[0] as scalar
• getLoadedPlugins() column names (⚠️ CRITICAL): getLoadedPlugins() returns a TABLE with EXACTLY these columns: [plugin, version, user, time].
  • Correct: exec plugin from getLoadedPlugins() where plugin = "HDF5"
  • WRONG: exec name from getLoadedPlugins() (no name column)
  • WRONG: exec pluginName from getLoadedPlugins() (no pluginName column)
  • WRONG: getLoadedPlugins().has_key("HDF5") (returns table not dict)
• HDF5 plugin: datasetName must be string scalar (⚠️ CRITICAL): hdf5::extractHDF5Schema(path, datasetName) and hdf5::loadHDF5(path, datasetName) require datasetName to be a STRING SCALAR, not a vector.
  • WRONG: dsName = exec tableName from datasets limit 1 (returns vector)
  • Correct: dsName = exec tableName from datasets limit 1; dsName = dsName[0] (take first element as scalar)
  • Correct: dsName = datasets.tableName[0] (directly take first element)
• dict() construction rules (⚠️ CRITICAL):
  • Syntax: dict(keyVector, valueVector) or dict(\key1`key2, [val1, val2])`
  • key and value count MUST be equal
  • WRONG: dict(["a","b"], [v1]) (count mismatch)
  • WRONG: dict(["a","b"], [exec count(*) from pt, exec sum(x) from pt]) (embedding exec SQL causes parser confusion)
  • Correct: rowCount = exec count(*) from pt; dict(["rows"], [rowCount]) (store in variable first)
  • WRONG: dict(["a","b"] as \a`b, ...)` (as syntax not applicable to dict)
• DolphinDB reserved words (⚠️ CRITICAL): DO NOT use these as variable names:
  • name, type, tuple, dict, table, select, exec, from, where, by, update, insert, delete
  • Use colName instead of name, dataTuple instead of tuple
  • In schema tables, name and type are column names, not variable names
• DolphinDB has NO .limit() method (⚠️ CRITICAL): Tables in DolphinDB do NOT have a .limit() method. To get top N rows:
  • Correct: select top 10 * from tableName
  • Correct: tableName[:10] (take first 10 rows)
  • WRONG: tableName.limit(10) (no such method)
  • WRONG: tableName.head(10) (no such method)
• DolphinDB function name pitfalls: DolphinDB built-in functions are LOWERCASE and case-sensitive. Common mistakes:
  • strLen → should be strlen
  • IsValid → should be isvalid (but isvalid doesn't exist, use other methods)
  • Count → should be count
  • Exists → should be existsDatabase/existsTable/existsShareVariable
  • valid() function does NOT exist — use t.size() > 0 or count(t) > 0 to check vector emptiness
• loadParquetEx parameter count (⚠️ CRITICAL): parquet::loadParquetEx accepts 4~9 arguments ONLY. Parameter order: (dbHandle, tableName, partitionColumns, fileName, [schema], [columnsToLoad], [startRowGroup], [rowGroupNum], [transform]). transform is the 9th (last) parameter. Do NOT pass more than 9 arguments.
• schema() function usage: schema(table) returns a schema object (NOT a table).
  • Correct: schema(pt) returns directly
  • Correct: schema(pt).colDefs gets column definition table
  • WRONG: select name, type from schema(pt) (schema() cannot be used in FROM clause)
• extractTextSchema returns metadata table, NOT schema table (⚠️ CRITICAL): extractTextSchema returns a table with columns [name, type, ...] describing the schema. To use it with createPartitionedTable, convert it to an empty table first: schemaTable = table(1:0, schema.name, schema.type). NEVER pass extractTextSchema result directly to createPartitionedTable.
• Derived partition column: When the source data has no suitable partition column but has a timestamp column, derive a DATE column from the timestamp for partitioning.
  • Example: addColumn(t, trade_date, DATE); t[trade_date] = date(exec timestamp from t)
  • Then use trade_date as the partition column
• File deletion in DolphinDB:
  • rmdir(path): deletes EMPTY directory only, NOT files
  • deleteFile(path): deletes a file (correct function for file deletion)
  • For temporary files, use try { deleteFile(path) } catch(ex) {} pattern
• Plugin function parse-time recognition: Plugin functions with :: prefix (e.g., parquet::extractParquetSchema, hdf5::lsTable) are recognized at PARSE time. If the plugin is not loaded before parsing, the function call will fail with "Cannot recognize function".
  • Solution 1: Use use pluginName after loadPlugin to import namespace, then call bare function names
  • Solution 2: Ensure loadPlugin is called BEFORE the script block containing :: functions
  • Solution 3: For compile probe, wrap plugin calls in try/catch or use use approach
• JSON serialization limitations: toStdJson() and internalToStdJson do NOT support all data forms:
  • Supported: STRING, INT, DOUBLE, BOOL, DATE, TIMESTAMP and other scalars and vectors
  • NOT supported: PAIR, MATRIX, DICT, SET and other complex data forms
  • Solution: convert complex types to simple vectors before serialization, or use string() conversion

Confirmation Strategy

Data import involves irreversible operations. Follow these confirmation rules:

Mandatory Confirmation (must stop and ask user before proceeding):

• If the user has not provided a file path: ask via ask_user to provide one. If the user's message does NOT contain a file path (no string matching patterns like /path/, .csv, .txt, .parquet, .h5, .bin), you MUST use ask_user to ask for the file path BEFORE doing anything else. Do NOT attempt to execute any DolphinDB scripts until you have a file path. This is the FIRST step of the workflow and cannot be skipped. If the user does not provide a file path, end the workflow gracefully.
• If the user has not specified the target form (memory table vs distributed table): ask via ask_user. Default is memory table.
• Memory table variable name confirmation: When importing to a memory table, you MUST ask the user to confirm the variable name (e.g., t in t = loadText(...)). Suggest a default name based on the file name (e.g., trades for trades.csv). Before assigning, check if the variable name already exists using:

  try { objByName("varName", false); varExists = true } catch(ex) { varExists = false }
  

  If the name exists, ask the user whether to overwrite (use undef(\varName)` then reassign) or rename. NEVER overwrite a variable without user confirmation.
• Before creating a distributed table: output engine selection, partition scheme, schema mapping, and confirmation question; wait for user confirmation before executing CREATE DATABASE/TABLE. NEVER assume a database path — always ask the user to confirm.
• Existing database/table handling (⚠️ CRITICAL): Before creating a distributed table, check both database and table existence. Use the following code pattern:

  dbExists = existsDatabase("dfs://path")
  tableExists = iif(dbExists, existsTable("dfs://path", "tableName"), false)
  

  ⚠️ API pitfall: existsTable("dfs://path", "tableName") and listTables("dfs://path") take a path string, NOT a database() object. db.listTables() will cause "db isn't an instance of a class" error. If the database exists: (i) If existsTable returns false (table name differs from any existing table), import directly into the existing database without dropping it. (ii) If existsTable returns true (table name matches an existing table), ask the user whether to overwrite (drop and recreate) or append. NEVER drop an existing database or table without explicit user confirmation.
• Before binary file import: output inferred schema and data preview; wait for user confirmation that the schema is correct before executing loadRecord or readRecord!.
• Schema confirmation is MANDATORY (Step 4): After parsing metadata, present the COMPLETE schema (with auto-detected time-type suggestions) to the user via ask_user and wait for confirmation. NEVER skip this step, even when all types appear correct and no conversion is needed — the user may want to adjust types, and their confirmation takes priority.

Conditional Confirmation (stop and ask when unclear):

• Storage engine is not specified and multiple reasonable choices exist.
• Partition scheme has multiple reasonable options — present specific choices based on data columns and selected engine (e.g., VALUE(time_col), HASH(category_col, N), COMPO(VALUE + HASH)). For IOTDB, only COMPO with time as LAST dimension.
• Column type is ambiguous: STRING columns containing delimited values that may be array vectors (e.g., "1.0,2.0,3.0"), or LONG/INT columns containing values that look like dates (8-digit integers) or timestamps (10-13 digit integers). Use ask_user to confirm the intended type and any conversion format.
• Date/time format cannot be inferred from data sample.
• Null handling strategy may affect business semantics.

Can Continue But Must Explain:

• File format inferred from extension — state the inference basis.
• Schema correction following skill rules (e.g., LONG timestamp → timestamp()) — state what was corrected.
• Import method auto-selected by file size — state the selection basis.

Quick Path

• When user instruction already specifies engine, partition scheme, or target table, skip corresponding confirmation points and proceed directly.
• When user requests quick import, use defaults (OLAP engine + VALUE partition) and skip non-mandatory confirmations; state which confirmations were skipped.
• When user provides the writeRecord script that created a binary file, derive schema from that script directly; no need to ask about schema.

Workflow

Follow these steps in order. For each step, if the information is already provided by the user or can be reliably inferred, proceed without asking. If confirmation is needed, use ask_user.

1. Confirm file path: Verify the data file path provided by the user. If the user's message does NOT contain a file path (no string matching patterns like /path/, .csv, .txt, .parquet, .h5, .bin), you MUST use ask_user to ask for the file path BEFORE doing anything else. Do NOT attempt to execute any DolphinDB scripts until you have a file path. This is the FIRST step of the workflow and cannot be skipped. If the user does not provide a file path, end the workflow gracefully.

2. Identify file format: Auto-detect format from file extension (.csv, .txt, .bin, .parquet, .h5/.hdf5) or file header. If ambiguous, ask the user.

3. Parse metadata: For text files, use extractTextSchema(filename, [delimiter]) to preview column names, types, and delimiter. For Parquet, use parquet::extractParquetSchema. For HDF5, use hdf5::extractHDF5Schema (after loading plugin and exploring with hdf5::lsTable). For binary files, there is no auto-detection function — the schema MUST be provided by the user (see Binary File Schema below).

4. Auto-detect time-type columns and present schema for user confirmation: Apply the Time-Type Auto-Detection rules below to identify columns that may need type conversion. Present the COMPLETE schema (including auto-detected type suggestions) to the user via ask_user. The user MUST confirm the schema before proceeding. This step is MANDATORY — never skip it, even when all types appear correct and no conversion is needed. The user may want to adjust types, and their confirmation takes priority.

5. Confirm target form: If the user has not specified the target form, ask: memory table or distributed table? Default is memory table if not specified. If memory table:

   • Ask user to confirm the variable name (e.g., t in t = loadText(...)). Suggest a default name based on the file name (e.g., trades for trades.csv).
   • Before assigning, check if the variable name already exists using:

     try { objByName("varName", false); varExists = true } catch(ex) { varExists = false }
     

     If the name exists, ask the user whether to overwrite (use undef(\varName)` then reassign) or rename. NEVER overwrite a variable without user confirmation. If distributed table:
   • Ask user to confirm database path and table name (suggest defaults based on file name). NEVER assume a database path without user confirmation.
   • Before creating, check both database and table existence using:

     dbExists = existsDatabase("dfs://path")
     tableExists = iif(dbExists, existsTable("dfs://path", "tableName"), false)
     

     ⚠️ API pitfall: existsTable and listTables take a path string (e.g., "dfs://sensor_db"), NOT a database() object. db.listTables() will cause "db isn't an instance of a class" error — always use listTables("dfs://path") instead. If the database exists: (i) If existsTable returns false (table name differs from any existing table), import directly into the existing database without dropping it. (ii) If existsTable returns true (table name matches an existing table), ask the user whether to overwrite (drop and recreate) or append via ask_user. NEVER drop an existing database or table without explicit user confirmation.
   • Ask user to confirm storage engine (OLAP/TSDB/PKEY/IOTDB) with recommended default.
   • Ask user to confirm partition scheme with specific options based on data columns.

6. Confirm import method: Based on file format and target, select the appropriate import function:

   • Text → memory table: loadText or ploadText
   • Text → distributed table: loadTextEx
   • Text → very large file: textChunkDS + mr
   • Binary with string columns: loadRecord
   • Binary without string columns: readRecord! (preferred)
   • Parquet → memory table: parquet::loadParquet
   • Parquet → distributed table (no sortColumns needed): parquet::loadParquetEx
   • Parquet → distributed table (sortColumns needed): parquet::loadParquet + replaceColumn! + pt.append!(data)
   • HDF5 → memory table: hdf5::loadHDF5
   • HDF5 → distributed table (no sortColumns needed): hdf5::loadHDF5Ex
   • HDF5 → distributed table (sortColumns needed): hdf5::loadHDF5 + replaceColumn! + pt.append!(data)

7. Data preprocessing: Apply any user-specified or default preprocessing before or during import:

   • Specify date/time format for date columns (via schema format column or replaceColumn! after import)
   • Select specific columns to import (via schema col column for text files, columnsToLoad for Parquet)
   • Skip rows (via skipRows for text files, skipBytes for binary files)
   • Handle null values (via transform parameter with nullFill!, ffill!, etc.)
   • For Parquet: specify startRowGroup and rowGroupNum to read specific row groups; specify columnsToLoad to load specific columns
   • For HDF5: use startRow and rowNum parameters in hdf5::loadHDF5/hdf5::loadHDF5Ex for efficient row range selection

8. Execute import: Run the import script. If errors occur, diagnose and fix (e.g., schema mismatch, partition conflict, plugin not loaded).

9. Verify result: Execute select count(*) from tableName (NOT tableName.count()) and select top 10 * from tableName. For distributed tables, tableName.count() may return 0 or incorrect results — always use SQL select count(*).

   ⚠️ Verification mode depends on import mode (determined by prompt content and table existence check — see "Existing database/table handling" and "Non-interactive mode" rules):

   Non-append import mode (DEFAULT — new table OR truncate-then-import): After import, simply verify exec count(*) from loadTable(dbPath, tableName) == expectedImportCount and check select top 10 * from loadTable(dbPath, tableName) for data correctness. Since the table was either newly created or cleared with truncate(dbPath, tableName) before import, there are NO pre-existing rows to conflict with — the final row count should exactly equal the source data row count for ALL storage engines (OLAP, TSDB, PKEY, IOTDB). No beforeCount/afterCount tracking or key-existence check is needed in this mode.

   Append import mode (⚠️ engine-aware verification REQUIRED — ONLY when prompt explicitly mentions "append"/"追加"): When data is appended to an existing table that already contains rows, the simple row-count equality check may FAIL for overwrite/deduplication engines. In append mode, MUST use the engine-aware verification mechanism based on the storage engine, because tableInsert/append!/loadTextEx return values are unreliable for PKEY/key-value tables (PKEY engine only counts NEW rows, not updates from key conflicts; append! and loadTextEx do not return inserted count at all):

   Strategy A — Row count difference check (non-overwrite engines): Use when the storage engine is OLAP, or TSDB with keepDuplicates=ALL (default). Every appended row is added without deduplication, so the row count difference accurately reflects imported rows:

   beforeCount = exec count(*) from loadTable(dbPath, tableName)
   // ... execute append import ...
   afterCount = exec count(*) from loadTable(dbPath, tableName)
   expectedImportCount = /* row count of source data */
   // Verify exact equality:
   if (afterCount - beforeCount == expectedImportCount) { /* success */ }
   

   Strategy B — Key existence check (overwrite/deduplication engines): Use when the storage engine is PKEY, or TSDB with keepDuplicates=LAST or keepDuplicates=FIRST. In append mode, the engine may OVERWRITE existing rows with the same primary key (PKEY) or KEEP only the last/first occurrence (TSDB+LAST/FIRST), so afterCount - beforeCount may be LESS than expectedImportCount — the row count difference equality check (Strategy A) WILL FAIL. Use three-step verification instead:

   // Step 1: record beforeCount BEFORE append import
   beforeCount = exec count(*) from loadTable(dbPath, tableName)
   // ... execute append import ...
   // Step 2: record afterCount AFTER append import
   afterCount = exec count(*) from loadTable(dbPath, tableName)
   // Step 3: extract unique key combinations from source data
   //   - PKEY engine: use the columns declared in `primaryKey`
   //   - TSDB engine with keepDuplicates=LAST/FIRST: use `sortColumns` columns
   //   - If sortColumns == partition column only, append the partition column to form a composite key
   uniqueKeys = select distinct keyCol1, keyCol2 from sourceData
   // Step 4: extract the same key combinations from the target table
   targetKeys = select distinct keyCol1, keyCol2 from loadTable(dbPath, tableName)
   // Step 5: use ej (inner join) to find keys present in BOTH uniqueKeys and targetKeys
   matchedKeys = ej(uniqueKeys, targetKeys, `keyCol1`keyCol2)
   

   Verify three conditions (all must hold):

   • B1 (lower bound): afterCount >= beforeCount — rows must not decrease after append import.
   • B2 (upper bound): afterCount - beforeCount <= expectedImportCount — appended rows must not exceed source row count.
   • B3 (key existence): exec count(*) from matchedKeys == exec count(*) from uniqueKeys — every unique key in the source data MUST exist in the target table.

   Determining key columns (Strategy B only):

   • PKEY engine: use the primaryKey parameter columns specified in createPartitionedTable.
   • TSDB engine with keepDuplicates=LAST/FIRST: use the sortColumns parameter columns. If sortColumns contains only one column that coincides with the partition column, include the partition column to form a composite key.

   ⚠️ Why Strategy A fails for PKEY/TSDB+LAST/FIRST in append mode: PKEY engine uses Merge-on-Write — when a new row has the same primary key as an existing row, the new row OVERWRITES the existing row (UPDATE), not an INSERT. The engine increments the row count only by the number of NEW primary keys, which may be less than expectedImportCount. Similarly, TSDB+LAST/FIRST deduplicates rows with the same sortColumns value, keeping only the last/first occurrence — repeated keys reduce the final row count. Therefore, afterCount - beforeCount no longer equals expectedImportCount, and the equality check in Strategy A is invalid. Use Strategy B (key existence check) instead.

   ⚠️ NEVER rely on tableInsert/append!/loadTextEx return values for verification in append mode: For PKEY engines and key-value tables, tableInsert return value only counts NEW rows inserted (NOT rows updated due to key conflicts); append! and loadTextEx do not return inserted count at all. Always use SQL select count(*) and key-existence checks instead.

10. Output summary: Report data source, row count, target table, storage engine, partition scheme, any type conversions applied, and warnings.

Time-Type Auto-Detection Rules

When parsing metadata, automatically identify columns that may be time-related and suggest type conversions. Present these suggestions to the user for confirmation in Step 4.

Detection by column name pattern (case-insensitive):

• Column name contains date, day, trade_day, trading_day → likely DATE or INT yyyyMMdd
• Column name contains time, timestamp, ts → likely TIME, TIMESTAMP, or LONG timestamp
• Column name contains datetime → likely DATETIME or TIMESTAMP

Detection by data type and value pattern:

• INT column with 8-digit values (e.g., 20240101) → likely yyyyMMdd date. Suggest: DATE via temporalParse(string(col), "yyyyMMdd")
• INT column with 6-digit values (e.g., 930000) → likely HHmmss time. Suggest: TIME via temporalParse(string(col), "HHmmss")
• INT column with 9-digit values (e.g., 93000000) → likely HHmmssSSS time. Suggest: TIME via temporalParse(string(col), "HHmmssSSS")
• LONG column with 10-digit values → likely second-level Unix timestamp. Suggest: TIMESTAMP via timestamp(col)
• LONG column with 13-digit values → likely millisecond-level Unix timestamp. Suggest: TIMESTAMP via timestamp(col)
• LONG column with 16-digit values → likely microsecond-level timestamp. Suggest: NANOTIMESTAMP via nanotimestamp(col)
• STRING column matching date patterns → use temporalParse with appropriate format

User intent takes priority: If the user explicitly specifies a type conversion (e.g., "ts列转换为TIMESTAMP"), always follow the user's specification. If the user's specification causes errors, provide a clear error message and suggest the correct approach.

Invalid date flexible conversion (⚠️ CRITICAL — never silently drop data): When temporalParse(string(col), "yyyyMMdd") returns NULL for some values, it means those values are invalid dates (e.g., January 32nd = 20240132). Instead of simply filtering out NULL rows (which causes data loss), try to interpret the invalid dates intelligently:

• If the pattern suggests the dates are sequential (e.g., 20240131, 20240132, 20240133), treat the excess day numbers as continuing into the next month: 20240132 → 2024.02.01, 20240133 → 2024.02.02, etc.
• Use this formula: date(2024.01.01) + (tradingDay - 20240101) where 20240101 is the base date and tradingDay is the INT value. This converts sequential day numbers to actual dates.
• Always present the conversion approach to the user for confirmation before proceeding.
• NEVER silently drop data rows due to invalid date conversion.

Presentation format for Step 4: Present the schema as a table with columns: Column Name | Original Type | Suggested Type | Conversion Method Mark auto-detected suggestions with [auto]. Let user confirm or modify.

Distributed Table Creation Checklist

When creating a distributed table, follow this checklist to avoid common errors:

1. Use createPartitionedTable (NOT createTable) for partitioned tables.
2. Schema table must match the final column types (after conversion), not the raw file types.
3. Partition column types must match the actual data types in the table (e.g., if tradingDay is converted to DATE, partition column must be DATE, not INT).
4. For TSDB engine, sortColumns is required. Use pt.append!(data) (NOT data.append!(pt)) when appending data.
5. For parquet::loadParquetEx and hdf5::loadHDF5Ex: these do NOT support sortColumns. Use loadParquet/loadHDF5 + replaceColumn! + pt.append!(data) instead.
6. Before importing, validate that date conversion will not produce NULLs in partition columns (e.g., invalid dates like 20240132 will cause temporalParse to return NULL, which cannot be used as partition key).
7. Partition range validation (⚠️ CRITICAL): Before creating a database with RANGE or VALUE partitioning on a date/time column, you MUST query min(partitionColumn) and max(partitionColumn) from the actual data to determine the correct partition boundaries. NEVER assume the data range — always verify with actual data. After import, ALWAYS verify data correctness using the engine-aware verification mechanism described in Step 9 (Strategy A row-count difference equality for OLAP / TSDB+ALL; Strategy B key-existence check for PKEY / TSDB+LAST/FIRST). If verification fails, investigate immediately (likely causes: partition range too narrow causing silent data drop, PKEY/TSDB deduplication reducing row count, or import errors). When the user specifies a partition scheme (e.g., VALUE(date)), you MUST use that scheme. If the data range exceeds the partition boundaries, expand the boundaries rather than switching to a different partition scheme.
8. Existing database/table check (⚠️ CRITICAL): Before creating, always check both database and table existence:

   dbExists = existsDatabase("dfs://path")
   tableExists = iif(dbExists, existsTable("dfs://path", "tableName"), false)
   

   • If dbExists=false: create new database and table.
   • If dbExists=true and tableExists=false: reuse existing database, create new table in it.
   • If dbExists=true and tableExists=true: ask user whether to overwrite (drop and recreate) or append. NEVER drop without user confirmation. ⚠️ API pitfall: existsTable("dfs://path", "tableName") takes path string, NOT database() object. db.listTables() causes "db isn't an instance of a class" error — use listTables("dfs://path") instead.

Binary File Schema

DolphinDB does NOT provide a function to auto-detect the structure of binary files. The schema (column names, data types, and string column lengths) MUST be obtained from one of the following sources:

1. User provides the writeRecord script that created the file — derive schema from the source table definition. This is the most reliable method. When schema is derived from the writeRecord script, proceed directly to import. Do NOT attempt to verify file size or record count — this wastes react rounds and is unnecessary when schema is already known.
2. User provides the schema directly — column names, data types, and string column lengths.
3. User provides partial information — e.g., column names only, or data types only. Use ask_user to fill in the gaps.

Why auto-inference is not feasible: A binary file is a raw byte sequence. The same file size can correspond to many different column type combinations (e.g., 8 bytes could be 1 LONG, or 2 INTs, or 1 DOUBLE). Without external metadata, it is impossible to reliably determine the schema. Even if the record count is known, the column count and types remain ambiguous. Therefore, ALWAYS ask the user for the schema before using loadRecord or readRecord!.

Getting file size in DolphinDB (only when truly needed): DolphinDB does NOT have getFileSize() or fileSize() functions. Use one of these two methods:

• file(path).seek(0, TAIL) — returns file size in bytes (recommended for single file).
• select fileSize from files(directory) where filename = "name" — returns file size from directory listing. Do NOT attempt non-existent functions like getFileSize(), fileSize(), or file(path).size().

React Round Conservation

The agent has a limited number of react rounds (tool calls). To avoid running out of rounds:

• Do NOT verify information that is already known. If the user provides the writeRecord script, derive schema directly and proceed — do NOT check file size or record count.
• Do NOT retry failed DolphinDB scripts with minor variations. If a function call fails (e.g., db.listTables() returns "db isn't an instance of a class"), STOP immediately and check the correct function signature in this skill's reference files. NEVER retry the same failing approach more than once — switch to the correct API (e.g., use listTables("dfs://path") instead of db.listTables()).
• Batch independent operations. If you need to check multiple things, combine them into a single DolphinDB script when possible.
• Prioritize the critical path. Focus on schema confirmation → import → verification. Skip optional verification steps if react rounds are limited.

Stop Conditions

• Source file does not exist or cannot be accessed.
• File format cannot be identified and user has not specified.
• Binary file schema cannot be determined and user has not provided.
• Database or table already exists and cannot be overwritten.
• Distributed table creation plan has not been confirmed by user.
• Storage engine is unclear and multiple reasonable choices exist.
• Date/time format cannot be inferred and affects data correctness.
• User's confirmation contradicts the previous plan.

Storage Engine Selection

Engine	Best For	Key Feature
OLAP	Bulk analysis, full-table scan	Append-only, highest write throughput
TSDB	Time-series point queries	sortColumns index, deduplication (keepDuplicates)
PKEY	Primary-key uniqueness, near-real-time updates	primaryKey constraint, bloom filter index
IOTDB	IoT massive point management	IOTANY variable-type column, latestKeyCache (ms-level latest-value query), static table mapping

IOTDB Engine Quick Reference (IoT Scenarios)

IOTDB is a storage engine based on TSDB, specifically designed for massive IoT point data management (available since DolphinDB 3.00.2).

Key concepts (Point, IOTANY, latestKeyCache, Static table) → reference/iotdb_engine.md

IOTDB creation rules (strict):

• Database engine must be IOTDB: database(..., engine='IOTDB')
• When creating a point management table, the database must use COMPO partition with time partition as the LAST dimension (required by IOTDB engine; single-layer partition will cause point table creation to fail)
• sortColumns must include all point-identifying columns (ID + tags) + time column as the LAST sort column
• If table contains IOTANY column, latestKeyCache=true is mandatory
• When latestKeyCache=true, sortColumns must have at least 2 columns
• Only point-management tables are allowed in an IOTDB database (no dimension tables)
• Deduplication: ALL (default) or LAST; FIRST is NOT supported
• IOTANY column type is fixed per point after first write; use update with firstSortKey-only WHERE to change

For complete creation examples, see reference/iotdb_engine.md and templates/iotdb_creation.md.

IOTDB Complete Creation Template (MUST follow this exact pattern):

// IOTDB engine: MUST use COMPO partition with time as LAST dimension
// Step 1: Create sub-databases for COMPO partition
hashPart = database("", HASH, [INT, 20])  // or [SYMBOL, 10] etc.
timePart = database("", VALUE, 2024.01.01..2024.12.31)

// Step 2: Create database with IOTDB engine
db = database("dfs://iot_db", COMPO, [hashPart, timePart], engine="IOTDB")

// Step 3: Create table with IOTANY column using SQL statement
// NOTE: IOTANY type can ONLY be specified in SQL CREATE statement, NOT in table() function
// NOTE: SQL create table statement is executed directly in DolphinDB script (NOT via db.run())
create table `dfs://iot_db`.`sensors` (
    deviceId INT,
    location SYMBOL,
    timestamp TIMESTAMP,
    value IOTANY
)
partitioned by deviceId, timestamp
sortColumns=`deviceId`location`timestamp
latestKeyCache=true

Key IOTDB rules:

• IOTDB engine MUST use COMPO partition with time column as the LAST partition dimension
• IOTANY type can ONLY be created via SQL create table statement, NOT via table() function or createPartitionedTable
• latestKeyCache=true is recommended for latest value queries
• sortColumns should include: id columns + tag columns + time column (time as LAST)
• COMPO partition sub-databases must be created separately using database("", ...) first

Latest-value query optimization:

• latestKeyCache=true → uses cache table (fastest, ms-level)
• Storage engine optimization → uses index zonemap filtering
• Check with select [HINT_EXPLAIN] ... context by ... limit -1

Common Import Patterns

// CSV/TXT → memory table (with schema correction)
schema = extractTextSchema("/data/trades.csv")    // ▶ 替换为你的文件路径
update schema set type = `SYMBOL where name = "ticker"
t = loadText("/data/trades.csv", schema=schema)

// CSV/TXT → distributed table (recommended)
db = database("dfs://stock", VALUE, 2024.01.01..2024.12.31)    // ▶ 替换为你的数据库路径和分区
loadTextEx(db, "trades", `date, "/data/trades.csv", sortColumns=`sym`date)

// CSV/TXT with transform (preprocess before writing)
loadTextEx(db, "trades", `date, "/data/trades.csv",
           transform = nullFill!{, 0.0}, sortColumns=`sym`date)

// Large file chunked import
ds = textChunkDS("/data/huge.csv", 500)    // ▶ 替换为你的文件路径
mr(ds, append!{pt}, ,, false)

// Long timestamp conversion
t = loadText("/data/ticks.csv")    // ▶ 替换为你的文件路径
replaceColumn!(t, `ts, timestamp(exec ts from t))

// Tab-separated TXT file
t = loadText("/data/demo.txt", delimiter=char(9))    // ▶ 替换为你的文件路径

All text import functions support both .csv and .txt. Default delimiter: comma. Tab-separated TXT: delimiter=char(9). DolphinDB does not support as an escape sequence; use char(9) to represent the Tab character.

For date/time cleaning, array vectors, null filling → reference/functions/data_cleaning.md. For IOTDB creation → reference/iotdb_engine.md and templates/iotdb_creation.md. For MQTT/OPC UA → reference/iot_ingestion.md and templates/mqtt_ingestion.md.

IoT Data Ingestion Architecture

Devices/Sensors → MQTT/OPC UA → DolphinDB Plugin → Stream Table → Stream Engine → Distributed Table

For details → reference/iot_ingestion.md and templates/mqtt_ingestion.md.

Bundled Files

Path	Purpose
reference/functions/*.md	Import function references (loadText, loadTextEx, textChunkDS, extractTextSchema, JSON, binary, data_cleaning, HDF5, Parquet)
reference/confirmation_templates.md	Confirmation output templates
reference/import_plugins.md	Database and file format plugins
reference/iotdb_engine.md	IOTDB engine detailed guide
reference/iot_ingestion.md	IoT real-time data ingestion
reference/iot_data_model.md	IoT data modeling
reference/partition_design.md	Partition strategy design
reference/engine_selection.md	Storage engine comparison
reference/pitfalls.md	Common import pitfalls
reference/long_timestamp.md	Long timestamp handling
templates/*.md	Step-by-step templates (CSV import, database creation, IOTDB creation, MQTT ingestion, batch import)
examples/*.dos	DolphinDB script examples — read and adapt the content, replace placeholder variables with actual values
checklists/import_review.md	Import result verification checklist

How to use bundled files: Read the relevant reference or template file within this skill package, then adapt the DolphinDB scripts to your actual scenario. For .dos examples, read the example content and replace placeholder variables (file paths, database names, table names) with your actual values before executing as DolphinDB scripts.