MultithreadedTableWriter

Syntax

ErrorCodeInfo insert(Object... args)

Parameters

• args: A variable-length argument (varargs) indicating the record to be inserted.

Details

Insert a single record. Return a class ErrorCodeInfo containing errorCode and errorInfo. If errorCode is not "", MultithreadedTableWriter has failed to insert the data, and errorInfo displays the error message.

The class ErrorCodeInfo provides methods hasError() and succeed() to check whether the data is written properly. hasError() returns true if an error occurred, false otherwise. succeed() returns true if the data is written successfully, false otherwise.

Examples

MultithreadedTableWriter multithreadedTableWriter = new MultithreadedTableWriter("localhost", 8848, "admin", "123456", "dfs://valuedb3", "pdatetest",
                false, false, null, 10000, 1,
                5, "id", new int[]{Vector.COMPRESS_LZ4, Vector.COMPRESS_LZ4, Vector.COMPRESS_DELTA});
ErrorCodeInfo pErrorInfo = multithreadedTableWriter_.insert(new Date(2022, 3, 23), "AAAAAAAB", 10000000L);

List<List<Entity>> getUnwrittenData()

Details

Return a nested list representing the data that has not been written to the server. Note that MultithreadedTableWriter will release these data resources after retrieving them.

Examples

List<List<Entity>> unwrittenData = multithreadedTableWriter_.getUnwrittenData();

ErrorCodeInfo insertUnwrittenData(List<List<Entity>> records)

Parameters

• records: Data that needs to be written again, obtained via the getUnwrittenData method.

Details

Insert unwritten data. The result is in the same format as insert. The difference is that insertUnwrittenData can insert multiple records at a time.

Examples

List<List<Entity>> unwrittenData = multithreadedTableWriter_.getUnwrittenData();
ErrorCodeInfo ret = multithreadedTableWriter_.insertUnwrittenData(unwrittenData);

Status getStatus()

Get the current running status of the MultithreadedTableWriter object.

Return Value: Status is a class inherited from ErrorCodeInfo.

• Status attributes:

  • isExiting: Whether the writing thread is exiting.
  • errorCode: Error code.
  • errorInfo: Error information.
  • sentRows: Total number of successfully sent records.
  • unsentRows: Total number of unsent records.
  • sendFailedRows: Total number of failed records.
  • threadStatus: List of writing thread statuses.
  • threadId: Thread ID.
  • sentRows: Number of successfully sent records by this thread.
  • unsentRows: Number of unsent records by this thread.
  • sendFailedRows: Number of failed records by this thread.

• Status methods:

  • hasError(): true indicates data write errors; false indicates no errors.
  • succeed(): true indicates data write success; false indicates failure.

Examples

The following example gets the status of an MTW object.

MultithreadedTableWriter.Status writeStatus = new MultithreadedTableWriter.Status();
writeStatus = multithreadedTableWriter_.getStatus();

The output of MultithreadedTableWriterStatus is shown below:

errorCode     : A1
errorInfo     : Data conversion error: Cannot convert double to LONG
isExiting     : True
sentRows      : 2493
unsentRows    : 0
sendFailedRows: 7507
threadStatus  :
        threadId        sentRows        unsentRows      sendFailedRows
               0               0                 0                7507
        3567691520           415                 0                   0
        3489658624           831                 0                   0
        3481265920           416                 0                   0
        3472873216           416                 0                   0
        3464480512           415                 0                   0
<dolphindb.session.MultithreadedTableWriterStatus object at 0x7f0102c76d30>

The output shows that an exception occurred during the MTW writing process.

• The error code is A1, with the exception information "Data conversion error: Cannot convert double to LONG."
• isExiting=True indicates that the current thread is exiting.
• sentRows=2493 shows that 2493 records have been successfully written to the DolphinDB server.
• unsentRows=0 and sendFailedRows=7507 indicate that a total of 7507 records are yet to be successfully written to the DolphinDB server.
• threadStatus lists the processing status of each background thread. threadId=0 represents the total count for all threads. Typically, the successful write results (sentRows) are shown separately in each thread, as seen in the sentRows column of the output above. The failed write results (unsentRows, sendFailedRows) are concentrated in the threadId=0 row, indicating the total number of failed writes throughout the process.

waitForThreadCompletion()

Details

This method puts the MTW into a waiting state, exiting once all background worker threads have completed their tasks. If you call insert or insertUnwrittenData after the execution of waitForThreadCompletion, an error "thread is exiting" will be raised.

Examples

The following example demonstrates how to wait for the Writer to complete writing.

multithreadedTableWriter_.waitForThreadCompletion();

If the data type of the inserted data does not match the column type of the table, MultithreadedTableWriter will immediately return an error message and print the stack trace:

DBConnection conn= new DBConnection();
conn.connect(HOST, PORT, "admin", "123456");
Random random = new Random();
String script =
        "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');";
conn.run(script);
MultithreadedTableWriter multithreadedTableWriter_ = new MultithreadedTableWriter(HOST, PORT, "admin", "123456", "dfs://valuedb3", "pdatetest",
        false, false, null, 10000, 1,
        5, "id", new int[]{Vector.COMPRESS_LZ4, Vector.COMPRESS_LZ4, Vector.COMPRESS_DELTA});
ErrorCodeInfo ret;
// Insert one row with incorrect data type, MTW returns an error message immediately
ret = multithreadedTableWriter_.insert(new Date(2022, 3, 23), 222, random.nextInt() % 10000);
if (!ret.errorInfo.equals(""))
    System.out.println("insert wrong format data: {2}\n" + ret.toString());

Execution result:

java.lang.RuntimeException: Failed to insert data. Cannot convert int to DT_SYMBOL.
	at com.xxdb.data.BasicEntityFactory.createScalar(BasicEntityFactory.java:795)
	at com.xxdb.data.BasicEntityFactory.createScalar(BasicEntityFactory.java:505)
	at com.xxdb.multithreadedtablewriter.MultithreadedTableWriter.insert(MultithreadedTableWriter.java:594)
	at com.xxdb.BehaviorTest.testMul(BehaviorTest.java:89)
	at com.xxdb.BehaviorTest.main(BehaviorTest.java:168)
insert wrong format data: {2}
code=A1 info=Invalid object error java.lang.RuntimeException: Failed to insert data. Cannot convert int to DT_SYMBOL.

If the number of columns in the inserted data does not match the number of columns in the table, MultithreadedTableWriter will immediately return an error message:

DBConnection conn= new DBConnection();
conn.connect(HOST, PORT, "admin", "123456");
Random random = new Random();
String script =
        "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');";
conn.run(script);
MultithreadedTableWriter multithreadedTableWriter_ = new MultithreadedTableWriter(HOST, PORT, "admin", "123456", "dfs://valuedb3", "pdatetest",
        false, false, null, 10000, 1,
        5, "id", new int[]{Vector.COMPRESS_LZ4, Vector.COMPRESS_LZ4, Vector.COMPRESS_DELTA});
ErrorCodeInfo ret;
// Insert one row of data where the number of columns does not match the table's columns, MTW returns an error message immediately
ret = multithreadedTableWriter_.insert(new Date(2022, 3, 23), random.nextInt() % 10000);
if (!ret.errorInfo.equals(""))
    System.out.println("insert wrong format data: {3}\n" + ret.toString());

Execution result:

insert wrong format data: {3}
code=A2 info=Column counts don't match.  

When MultithreadedTableWriter encounters errors such as mismatch errors or writing thread failures, the API will terminate all threads. You can use getUnwrittenData() to obtain the unwritten data and then rewrite it with insertUnwrittenData(). Please note that a new MTW object must be created to write the unwritten data.

List<List<Entity>> unwriterdata = new ArrayList<>();
unwriterdata = multithreadedTableWriter_.getUnwrittenData();
System.out.println("{5} unwriterdata: " + unwriterdata.size());
// Retrieve a new MTW object
MultithreadedTableWriter newmultithreadedTableWriter = new MultithreadedTableWriter(HOST, PORT, "admin", "123456", "dfs://valuedb3", "pdatetest",
        false, false, null, 10000, 1,
        5, "id", new int[]{Vector.COMPRESS_LZ4, Vector.COMPRESS_LZ4, Vector.COMPRESS_DELTA});
try
{
    boolean writesuccess = true;
    // Write the unwritten data to the new MTW
    ret = newmultithreadedTableWriter.insertUnwrittenData(unwriterdata);
}
finally
{
    newmultithreadedTableWriter.waitForThreadCompletion();
    writeStatus = newmultithreadedTableWriter.getStatus();
    System.out.println("writeStatus: {6}\n" + writeStatus.toString());
}

Execution result:

  {5} unwriterdata: 10
  writeStatus: {6}
  errorCode     : 
  errorInfo     : 
  isExiting     : true
  sentRows      : 10
  unsentRows    : 0
  sendFailedRows: 0
  threadStatus  :
          threadId        sentRows      unsentRows  sendFailedRows
                23               3               0               0
                24               2               0               0
                25               1               0               0
                26               3               0               0
                27               1               0               0