MultithreadedTableWriter

You can insert a single record with the insert function.

The function declaration is as follows:

template<typename... TArgs>
bool insert(ErrorCodeInfo &errorInfo, TArgs... args)

Arguments

• errorInfo: Out-parameter. An ErrorCodeInfo object, including error code and info, for the failed insert operation.
• args: Variadic arguments. The record to be inserted.
  Note:

  • If callbackFunc is specified when constructing the MTW, the first argument of args must be of string type that indicates the ID of the inserted row.
  • The insert function will release the pointer contained in args after execution, so it is recommended to pass ConstantSP or C++ basic type arguments.

Return Values

A boolean value indicating whether data is inserted successfully.

You can use the getUnwrittenData to obtain a list of data that has not been written to the server.

Function declaration is as follows:

void getUnwrittenData(std::vector<std::vector<ConstantSP>*> &unwrittenData);

Arguments

• unwrittenData: Out-parameter. A nested list of records that have not yet been written to the DolphinDB server. This includes records that failed to be sent and records yet to be sent.

Note: Calling getUnwrittenData() will remove those records from the MTW, releasing them from memory.

You can insert multiple unwritten records into the table with the insertUnwrittenData function.

The function declaration is as follows:

bool insertUnwrittenData(std::vector<std::vector<ConstantSP>*> &records, ErrorCodeInfo &errorInfo)

Arguments

• records: The records that have not yet been written to the DolphinDB server. Often obtained by calling getUnwrittenData().
• errorInfo: Out-parameter. An ErrorCodeInfo object, including error code and info, for the failed insert operation.

You can call the getStatus function to check the execution status of the MTW object.

The function declaration is as follows:

void getStatus(Status &status);

Arguments

• status: Out-parameter. A MultithreadedTableWriter::Status object.

Return Values

Returns a MultithreadedTableWriter::Status object, providing the following details:

• isExiting: Whether the writer thread(s) are exiting.
• errorCode: The error code. Empty string if no error.
• errorInfo: The error message. None if no error.
• sentRows: The number of records successfully sent (written) to server.
• unsentRows: The number of records to be sent, including the records pending conversion and the converted records pending insertion into the sending queue.
• sendFailedRows: The number of records failed to be sent (including records in the sending queue)
• threadStatus: a list of status for each writer thread, including:
  • threadId: The thread ID.
  • sentRows: The number of records this thread has sent.
  • unsentRows: The number of records pending to be sent by this thread.
  • sendFailedRows: The number of records this thread has failed to send.

Use the waitForThreadCompletion function to make the MTW wait until all working threads complete their tasks.

The function declaration is as follows:

void waitForThreadCompletion();

Arguments

None

int main(int argc, const char **argv)
{
    DBConnection conn;
    conn.connect("127.0.0.1", 8848, "admin", "123456");
    // create a table
    string script = "t = table(1000:0, `col1`col2`col3, [LONG, DOUBLE, STRING]); share t as t1;";
    conn.run(script);
    
    // construct an MTW object
    ErrorCodeInfo errorInfo;
    MultithreadedTableWriter writer("127.0.0.1", 8848, "admin", "123456", "", "t1", false);
    char msg[] = "123456msg";
    if(!writer.insert(errorInfo, 1, 2.3, msg)){
        std::cout << "insert fail " << errorInfo.errorInfo << std::endl;
        return -1;
    }
    writer.waitForThreadCompletion();
    // get the status of MTW
    MultithreadedTableWriter::Status status;
    writer.getStatus(status);
    if (status.hasError()) {
        std::cout << "error in writing: " << status.errorInfo << std::endl;
    }
    // check the results
    std::cout << conn.run("t1")->getString() << std::endl;
}

int main(int argc, const char **argv)
{
    DBConnection conn;
    conn.connect("127.0.0.1", 8848, "admin", "123456");
    // create a table
    string script = "t = table(1000:0, `col1`col2`col3, [LONG, DOUBLE, STRING]); share t as t1;";
    conn.run(script);

    auto cb = [](ConstantSP callbackTable)
	{
		/***
			callbackTable schema:
			column 0: id->string
			column 1: success->bool
		***/
		std::cout << callbackTable->getString() << std::endl;
	};

    // construct an MTW object
    ErrorCodeInfo errorInfo;
    MultithreadedTableWriter writer("127.0.0.1", 8848, "admin", "123456", "", "t1", false, false, nullptr, 1, 0.1, 5, "col1", nullptr, MultithreadedTableWriter::M_Append, nullptr, cb);
    char msg[] = "123456msg";
    if(!writer.insert(errorInfo, "row1", 1, 2.3, msg)){
        std::cout << "insert fail " << errorInfo.errorInfo << std::endl;
        return -1;
    }
    writer.waitForThreadCompletion();
    // get the status of MTW
    MultithreadedTableWriter::Status status;
    writer.getStatus(status);
    if (status.hasError()) {
        std::cout << "error in writing: " << status.errorInfo << std::endl;
    }
    // check the results
    std::cout << conn.run("t1")->getString() << std::endl;
}

int main(int argc, const char **argv)
{
    DBConnection conn;
	conn.connect("127.0.0.1", 8848, "admin", "123456");
	conn.run(R"(dbName = 'dfs://valuedb3'
                if(exists(dbName)){
                    dropDatabase(dbName);
                }
                datetest=table(1000:0,`date`symbol`id,[DATE,SYMBOL,LONG]);
                db = database(directory=dbName, partitionType=HASH, partitionScheme=[INT, 10]);
                pt=db.createPartitionedTable(datetest,'pdatetest','id');)");

    vector<COMPRESS_METHOD> compress;
    compress.push_back(COMPRESS_LZ4);
    compress.push_back(COMPRESS_LZ4);
    compress.push_back(COMPRESS_DELTA);
    MultithreadedTableWriter writer("127.0.0.1", 8848, "admin", "123456", "dfs://valuedb3", "pdatetest", false, false, NULL, 10000, 1, 5, "id", &compress);
    
    ErrorCodeInfo errorInfo;
    // insert 100 rows of records with correct data types and column count
    for (int i = 0; i < 100; i++) {
        if (writer.insert(errorInfo, rand() % 10000, "AAAAAAAB", rand() % 10000) == false) {
            // this line will not be executed
            cout << "insert failed: " << errorInfo.errorInfo << endl;
            break;
        }
    }
    // An error message is returned if the inserted row has wrong data type
    if (writer.insert(errorInfo, rand() % 10000, 222, rand() % 10000) == false) {
        cout << "insert failed: " << errorInfo.errorInfo << endl;// insert failed: Column counts don't match 2
    }
    // An error message is returned if the inserted row has wrong columns
    if (writer.insert(errorInfo, rand() % 10000, "AAAAAAAB") == false) {
        cout << "insert failed: " << errorInfo.errorInfo << endl;
    }
    writer.waitForThreadCompletion();
    cout << conn.run("select count(*) from pt")->getString() << endl;
}

int main(){
    // Establishes a connection and verifies the server status based on your own situation.
    std::shared_ptr conn = std::make_shared<DBConnection>();
    conn->connect("localhost", 8848, "admin", "123456", "", false, {}, 7200, true);
    //  Handles the connection based on your own situation.
    auto cb = [&conn](const MTWState state, const std::string &host, const int port) {
        std::cout << "This is a callback: state " << static_cast<int>(state) << ", server " << host << ":" << port << std::endl;
        return true;
    };
    // Constructs and configures the MTW class.
    MTWConfig config(conn, "myTable");
    config.onConnectionStateChange(cb).setThreads(5, "myColumn");
    MultithreadedTableWriter mtw(config);

    // Subsequent operations are independent of the constructor interface. Supports all MTW functions.
    ErrorCodeInfo pErrorInfo;
    string sym[] = {"A", "B", "C", "D"};
    for (int i = 0; i < 10; i++) {
        mtw.insert(pErrorInfo, sym[i % 4], i * 12, i + 64);
        std::this_thread::sleep_for(100ms);
    }
    mtw.waitForThreadCompletion();
    return 0;
}