mysql

Syntax

connect(host, port, user, password, db, [config])

Parameters

host: A string specifying the address of the MySQL server.

port: An int specifying the port of the MySQL server.

user: A string specifying the user name of the MySQL server.

password: A string specifying the password of the MySQL server.

db: A string specifying the database name.

config (optional): A dictionary specifying connection settings. The keys are strings, representing configuration names, and the values are the configuration settings.

• "SSL_ENFORCE": A boolean value specifying whether to enable SSL. It defaults to true.
• "SSL_VERIFY_SERVER_CERT": A boolean value specifying whether to verify the server certificate. It defaults to false.
• "CHARSET": A string specifying the character set used for the connection. It defaults to "UTF8".

Details

Create a connection to the MySQL server. Return a handle of MySQL connection, which will be used for operations including load and loadEx.

Example

conn = mysql::connect(`localhost, 3306, `root, `root, `DolphinDB)

Syntax

showTables(conn)

Parameters

• conn: A MySQL connection handle created with connect.

Details

List all table names in a MySQL database specified in connect.

Examples

conn = mysql::connect(`localhost, 3306, `root, `root, `DolphinDB)
mysql::showTables(conn)

/* output:
  Tables_in_DolphinDB
  -------------------
  US
*/  

Syntax

extractSchema(conn, tableName)

Parameters

• conn: A MySQL connection handle created with connect.
• tableName: A string indicating the name of a table in MySQL server.

Details

Generate the schema of a table.

Examples

conn = mysql::connect(`localhost, 3306, `root, `root, `DolphinDB)
mysql::extractSchema(conn, `US)

/* output:
        name    type        DolphinDBType
        PERMNO  INT         int(11)
        date    DATE        date
        SHRCD   INT         int(11)
        TICKER  SYMBOL      varchar(10)
        ...
        PRC     DOUBLE      double
*/        

Syntax

load(conn, table|query, [schema], [startRow], [rowNum], [allowEmptyTable])

Parameters

• conn: A MySQL connection handle created with connect.
• table|query: A string indicating the name of a MySQL server table or a valid MySQL query such as select * from table limit 100.
• schema: A table containing two STRING type columns. The first represents the column names used to create the result table, and the second represents the target data types. schema can contain additional columns as long as the first two meet the requirements. If we need to change the data type of a column that is automatically determined by the system, the schema table needs to be modified and used as a parameter. schema can be created manually or obtained through the extractSchema method.
• startRow: An integer indicating the index of the starting row to read. If unspecified, read from the first row. If 'table|query' is a SQL query, then 'startRow' should be unspecified.
• rowNum: An integer indicating the number of rows to read. If unspecified, read to the last row. If 'table_or_query' is a SQL query, then 'rowNum' should be unspecified.
• allowEmptyTable: A Boolean indicating whether to allow importing an empty table from MySQL The default value is false.This parameter is used to manage the loading restrictions on empty tables.

Details

Load a MySQL table or SQL query result into a DolphinDB in-memory table.

For details about supported data types as well as data conversion rules, please refer to the section of Data Types below.

Examples

conn = mysql::connect(`192.168.1.18, 3306, `root, `root, `DolphinDB)
tb = mysql::load(conn, `US,,0,123456)
select count(*) from tb

conn = mysql::connect(`127.0.0.1, 3306, `root, `root, `DolphinDB)
tb = mysql::load(conn, "SELECT PERMNO FROM US LIMIT 123456")
select count(*) from tb

mysql::load(conn, "SELECT now(6)");

Syntax

loadEx(conn, dbHandle, tableName, partitionColumns, table|query, [schema], [startRow], [rowNum], [transform], [sortColumns], [keepDuplicates], [sortKeyMappingFunction])

Parameters

• conn: A MySQL connection handle created with connect.
• dbHandle and tableName: If the input data is to be saved into a distributed database, the database handle and table name should be specified.
• partitionColumns: A STRING scalar/vector indicating partitioning column(s).
• table|query: A string indicating the name of a MySQL server table or a valid MySQL query such as select * from table limit 100. Note that the column order of the queried MySQL table is consistent with that of the DolphinDB distributed table; otherwise, value errors or type conversion failures may occur.
• schema: A table containing two STRING type columns. The first represents the column names used to create the result table, and the second represents the target data types. schema can contain additional columns as long as the first two meet the requirements. If we need to change the data type of a column that is automatically determined by the system, the schema table needs to be modified and used as a parameter. schema can be created manually or obtained through the extractSchema method.
• startRow: An integer indicating the index of the starting row to read. If unspecified, read from the first row. If 'table|query' is a SQL query, then 'startRow' should unspecified.
• rowNum: An integer indicating the number of rows to read. If unspecified, read to the last row. If 'table|query' is a SQL query, then 'rowNum' should unspecified.
• transform: Apply certain transformation on a MySQL table or query before importing into DolphinDB database.
• sortColumns: A string scalar or vector used to specify the sorting column of the table. The written data will be sorted according to sortColumns. This is only required when creating a table with the TSDB engine.
• keepDuplicates: Specifies how to handle data with identical values for all sortColumns within each partition. "ALL" is used to retain all data and is the default value. "LAST" retains only the most recent data. "FIRST" retains only the first piece of data. This is only required when creating a table with the TSDB engine.
• sortKeyMappingFunction: A vector composed of unary function objects, with a length consistent with the index columns, that is, the length of sortColumns minus 1. It specifies the mapping functions to be applied to each column in the index columns in order to reduce the number of sort key combinations. This process is called sort key dimensionality reduction. This is only required when creating a table with the TSDB engine.

Details

Load a MySQL table as a distributed table. The result is a table object with loaded metadata.

For details about supported data types as well as data conversion rules, please refer to the Data Types section.

Examples

• Load data as a partitioned table on disk.

  dbPath = "C:/..."
  db = database(dbPath, RANGE, 0 500 1000)
  mysql::loadEx(conn, db,`tb, `PERMNO, `US)
  tb = loadTable(dbPath, `tb)

• Load data as an in-memory partitioned table

  Load the entire table

  db = database("", RANGE, 0 50000 10000)
  tb = mysql::loadEx(conn, db,`tb, `PERMNO, `US)

  Load via SQL statement

  db = database("", RANGE, 0 50000 10000)
  tb = mysql::loadEx(conn, db,`tb, `PERMNO, "SELECT * FROM US LIMIT 100");

• Load data as a DFS partitioned table

  Load the entire table

  db = database("dfs://US", RANGE, 0 50000 10000)
  mysql::loadEx(conn, db,`tb, `PERMNO, `US)
  tb = loadTable("dfs://US", `tb)

  Load via SQL statement

  db = database("dfs://US", RANGE, 0 50000 10000)
  mysql::loadEx(conn, db,`tb, `PERMNO, "SELECT * FROM US LIMIT 1000");
  tb = loadTable("dfs://US", `tb)

  Load and transform data into a DFS partitioned table

  db = database("dfs://US", RANGE, 0 50000 10000)
  def replaceTable(mutable t){
  	return t.replaceColumn!(`svalue,t[`savlue]-1)
  }
  t = mysql::loadEx(conn, db, "",`stockid, 'select  * from US where stockid<=1000000',,,,replaceTable)

  Load via the schema parameter

  // Obtain the schema of the example table using extractSchema and convert the type of a specific column to DOUBLE
  schema = extractSchema(conn, "example")
  update schema set type = "double" where name = "column"
  
  // Create a database and load the example table to the pt table in the dfs://example database
  db = database("dfs://example", RANGE, 0 50000 1000000 1500000 2000000 2500000 3000001)
  mysql::loadEx(conn, db, `pt, `partitionColumn, "example", schema)

Syntax

mysql::subscribeBinlog(conn, sourceTables, targetTable, [options])

Details

Creates a Binlog subscription that captures INSERT, UPDATE, and DELETE operations on specified MySQL tables in real time, and asynchronously appends the MySQL change data capture (CDC) records to a DolphinDB table. If a valid checkpoint exists, the subscription resumes from the checkpoint; otherwise, it starts from the current MySQL Binlog position at the time of the call.

Before using this method, establish a connection with mysql::connect. The MySQL server must have log_bin=ON, binlog_format=ROW, and binlog_row_image=FULL configured. The connecting user must have permission to read the Binlog, query the current Binlog position, and access the source table schemas. The source tables must not contain columns of the native MySQL JSON type.

After the subscription is created successfully, the passed conn can continue to be used for other queries or be closed separately. However, mysql::close(conn) and mysql::subscribeBinlog(conn, ...) must not execute concurrently. If the source table schema changes while the subscription is active, stop the subscription and recreate the subscription.

Parameters

conn is an open MySQL connection handle created by mysql::connect. The method uses this connection's configuration to create separate metadata and replication connections.

sourceTables is a table containing information about the source tables to subscribe to. The table must not be empty, and the same source table must not appear more than once. The following two columns are required:

targetTable is the DolphinDB table that receives CDC data. The table must contain exactly the following 9 columns. Their names, order, and types must not be changed:

The row data corresponding to each operation is as follows:

Integers outside the JSON safe integer range and DECIMAL values are represented as JSON strings in before and after; binary, BIT, and GEOMETRY data are represented as 0x...

options (optional) is a dictionary of type (STRING/SYMBOL -> ANY). The following configuration options are supported:

The starting position cannot be specified manually using binlogFile or binlogPosition.

checkpointTable must use subscriptionName as its first partition column and contain exactly the following 7 columns:

To resume a subscription, you must use the same subscriptionName, source tables, and target table. The plugin resumes from the record with the highest checkpointSeq and uses (binlogFile, eventPos, rowIndex) to avoid rewriting CDC records that already exist.

Returns

MySQL Binlog subscription resource. You should retain this resource; the subscription stops automatically when the resource is released.

Example

conn = mysql::connect("127.0.0.1", 3306, "test_user", "123456", "test2")

sourceTables = table(
    ["test2"] as databaseName,
    ["binlog_case"] as tableName
)

targetTable = table(
    1000:0,
    `databaseName`tableName`operation`eventTime`binlogFile`eventPos`rowIndex`before`after,
    [SYMBOL, SYMBOL, SYMBOL, TIMESTAMP, STRING, LONG, INT, STRING, STRING]
)

options = dict(STRING, ANY)
options["subscriptionName"] = "mysql_binlog_case"

subscription = mysql::subscribeBinlog(conn, sourceTables, targetTable, options)

// After the subscription starts successfully, execute an INSERT, UPDATE, or DELETE on test2.binlog_case in MySQL.
select * from targetTable

Syntax

mysql::unsubscribeBinlog(subscription)

Details

Stops the specified subscription, waits for the background thread to exit, and releases the connection used by the subscription. Once stopped, the subscription no longer writes data to the CDC or checkpoint table. It does not close the conn passed when the subscription was created. You should also call this method for subscriptions in the FAILED state to release their resources.

Parameters

subscription can be either a subscription resource returned by subscribeBinlog or a non-empty STRING scalar containing the subscription name. To stop a subscription by name, the caller must be the subscription owner or an administrator.

Example

mysql::unsubscribeBinlog(subscription)

// You can also stop the subscription by name.
mysql::unsubscribeBinlog("mysql_binlog_case")

Syntax

mysql::getBinlogStatus([subscription])

Details

Queries the status and processing progress of Binlog subscriptions. When a subscription resource is specified, the function returns one row containing its status. When no argument is specified, it returns all unreleased subscriptions visible to the current user. Administrators can view all such subscriptions. After a subscription is stopped, you can still use its resource to query its final status, provided that the resource has not been released.

Parameters

subscription (optional) is a subscription resource returned by subscribeBinlog.

Returns

A status table with the following columns:

Example

// Query the specified subscription.
status = mysql::getBinlogStatus(subscription)

// Query unreleased subscriptions visible to the current user.
allStatus = mysql::getBinlogStatus()

select subscriptionName, state, rowsWritten, lastError from status

Syntax

close(conn)

Parameters

conn: A MySQL connection handle created with connect.

Details

Disconnect and close the MySQL handle.

Example

mysql::close(conn)

• The numeric types in DolphinDB are all signed types. To prevent overflow, all unsigned types are converted to high-order signed types. For example, unsigned CHAR is converted to signed SHORT, unsigned SHORT is converted to signed INT, etc. 64-bit unsigned types are not supported.
• 'unsigned long long' is not supported in DolphinDB, you can specify schema and use DOUBLE or FLOAT if needed.
• The smallest value of each integral type in DolphinDB is null value, e.g. -128 for CHAR, -32,768 for SHORT, -2,147,483,648 for INT and -9,223,372,036,854,775,808 for LONG all mean null values in each type respectively.

Note:

• IEEE754 floating-point types are all signed numbers.
• Floating-point types float and double can be converted to numeric types (BOOL, CHAR, SHORT, INT, LONG, FLOAT, DOUBLE) in DolphinDB.
• The newdecimal/decimal type can only be converted to DOUBLE.

• All data types above can be converted to temporal data types in DolphinDB (DATE, MONTH, TIME, MINUTE, SECOND, DATETIME, TIMESTAMP, NANOTIME, NANOTIMESTAMP).

• char and varchar types of length less or equal to 10 will be converted to SYMBOL type in DolphinDB. Other string types will be converted to STRING type in DolphinDB.
• string type will be converted to STRING or SYMBOL type in DolphinDB.

enum type will be converted to SYMBOL type in DolphinDB.

• CPU: i7-7700 3.60GHZ.
• Hard disk: SSD, read speed 460~500MB/s.

US stocks daily data from 1990 to 2016 with 22 fields and 50,591,907 rows. Total size is 6.5GB. Time consumed: 160.5 seconds.