我正在为DolphinDB开发一个DuckDB插件，并且有以下测试用例：

@testing:case="test_duckdb_load_to_dfs_with_transform"
-- A. Prepare DFS table
dbPath = "dfs://duck_test_db"
if(existsDatabase(dbPath)) dropDatabase(dbPath)
db = database(dbPath, VALUE, 2023.01.01..2023.01.10)
dummy = table(1:0, `ts`val, [TIMESTAMP, DOUBLE])
pt = db.createPartitionedTable(dummy, "pt", "ts")

-- B. Prepare data source (generate 100,000 rows in DuckDB in-memory table)
conn = duckdb::connect("")
duckdb::execute(conn, "CREATE TABLE source AS SELECT
CAST('2023-01-01 00:00:00' AS TIMESTAMP) + INTERVAL (range) SECOND as ts,
CAST(random() AS DOUBLE) as val
FROM range(100000)")

-- C. Define Transform function (add 100 to val)
def myTrans(mutable t) {
    return select ts, val+100 from t
}

-- D. Execute load
rowCount = duckdb::load(conn, "SELECT * FROM source", pt, 10000, myTrans)    
assert 1, rowCount == 100000
assert 2, (exec count(*) from pt) == 100000

在执行代码时，我收到了以下错误：

rowCount = duckdb::load(conn, "SELECT * FROM source", pt, 10000, myTrans) => Append Error: Can't append data to a segmented table that contains external partitions.

我的C++插件函数代码如下：

extern "C" __declspec(dllexport) ConstantSP duckdbLoad(Heap* heap, vector<ConstantSP>& args) {
    if (args.size() < 3) throw IllegalArgumentException("duckdb::load", "Usage: load(conn, sql, destTable, [batchSize], [transform])");

    DuckDBConn* wrapper = (DuckDBConn*)args[0].get();
    duckdb_connection conn = wrapper->getConn();
    string queryOrTable = args[1]->getString();
    TableSP destTable = (TableSP)args[2];
    int batchSize = (args.size() >= 4 && !args[3]->isNull()) ? args[3]->getInt() : 65536;
    FunctionDefSP transform = (args.size() >= 5 && !args[4]->isNull()) ? (FunctionDefSP)args[4] : nullptr;

    string sql = queryOrTable;
    if (queryOrTable.find_first_of(" \t\n\r") == string::npos) {
        sql = "SELECT * FROM " + queryOrTable;
    }

    duckdb_prepared_statement stmt;
    if (duckdb_prepare(conn, sql.c_str(), &stmt) == DuckDBError) {
        string err = duckdb_prepare_error(stmt);
        duckdb_destroy_prepare(&stmt);
        throw RuntimeException(err);
    }

    duckdb_result result;
    if (duckdb_execute_prepared(stmt, &result) == DuckDBError) {
        string err = duckdb_result_error(&result);
        duckdb_destroy_result(&result);
        duckdb_destroy_prepare(&stmt);
        throw RuntimeException(err);
    }

    int colCount = (int)duckdb_column_count(&result);
    vector<string> colNames;
    for(int i=0; i<colCount; ++i) colNames.push_back(duckdb_column_name(&result, i));

    auto createBatchCols = [&](int size) {
        vector<ConstantSP> cols(colCount);
        for(int i=0; i<colCount; ++i) {
            cols[i] = Util::createVector(destTable->getColumnType(i), 0, size);
        }
        return cols;
    };

    vector<ConstantSP> batchCols = createBatchCols(batchSize);
    long long totalInserted = 0;
    int accumulatedRows = 0;

    while (true) {
        duckdb_data_chunk chunk = duckdb_fetch_chunk(result);
        if (!chunk) break;
        int chunkRows = (int)duckdb_data_chunk_get_size(chunk);
        if (chunkRows == 0) {
            duckdb_destroy_data_chunk(&chunk);
            break;
        }

        for (int i = 0; i < colCount; ++i) {
            Vector* v = (Vector*)batchCols[i].get();
            v->resize(v->size() + chunkRows);
            fillDolphinVector(duckdb_data_chunk_get_vector(chunk, i), v, (idx_t)chunkRows);
        }
        accumulatedRows += chunkRows;

        if (accumulatedRows >= batchSize) {
            TableSP batchTable = Util::createTable(colNames, batchCols);
            if (!transform.isNull()) {
                vector<ConstantSP> tArgs = {batchTable};
                batchTable = transform->call(heap, tArgs);
            }

            vector<ConstantSP> toAppend;
            for(int i=0; i<batchTable->columns(); ++i) toAppend.push_back(batchTable->getColumn(i));

            int inserted; string errMsg;
            if (!destTable->append(toAppend, inserted, errMsg)) {
                duckdb_destroy_data_chunk(&chunk);
                duckdb_destroy_result(&result);
                duckdb_destroy_prepare(&stmt);
                throw RuntimeException("Append Error: " + errMsg);
            }
            totalInserted += inserted;
            accumulatedRows = 0;
            batchCols = createBatchCols(batchSize);
        }
        duckdb_destroy_data_chunk(&chunk);
    }

    if (accumulatedRows > 0) {
        TableSP batchTable = Util::createTable(colNames, batchCols);
        if (!transform.isNull()) {
            vector<ConstantSP> tArgs = {batchTable};
            batchTable = transform->call(heap, tArgs);
        }
        vector<ConstantSP> toAppend;
        for(int i=0; i<batchTable->columns(); ++i) toAppend.push_back(batchTable->getColumn(i));
        int inserted; string errMsg;
        if (destTable->append(toAppend, inserted, errMsg)) {
            totalInserted += inserted;
        }
    }

    duckdb_destroy_result(&result);
    duckdb_destroy_prepare(&stmt);
    return new Long(totalInserted);
}

我确定错误来自 destTable->append(toAppend, inserted, errMsg) 这一行，因此我也尝试将其替换为：

destTable->append(heap, toAppend, inserted, errMsg)

但没有效果。应该如何修改代码来解决这个问题？

解决方案

建议使用系统内置的append! 函数来向DFS表写入数据。消息中提到的“external partitions”错误，是DolphinDB存储引擎内核的一种自保护机制。当你直接调用分布式表的append方法时，内核会将其解读为在不涉及事务协调者的情况下绕过文件锁的尝试。通过调用append! 函数，实质上是在C++中执行脚本语句append!(destTable, tmpTable)，从而确保触发DolphinDB的完整事务生命周期。

示例代码如下：

// Obtain the system's built-in append! function definition
static FunctionDefSP appendFunc = heap->currentSession()->getFunctionDef("append!");

while (true) {
    duckdb_data_chunk chunk = duckdb_fetch_chunk(result);
    if (!chunk) break;
    int chunkRows = (int)duckdb_data_chunk_get_size(chunk);
    if (chunkRows == 0) {
        duckdb_destroy_data_chunk(&chunk);
        break;
    }

    for (int i = 0; i < colCount; ++i) {
        Vector* v = (Vector*)batchCols[i].get();
        v->resize(v->size() + chunkRows);
        fillDolphinVector(duckdb_data_chunk_get_vector(chunk, i), v, (idx_t)chunkRows);
    }
    accumulatedRows += chunkRows;

    if (accumulatedRows >= batchSize) {
        TableSP batchTable = Util::createTable(colNames, batchCols);

        if (!transform.isNull()) {
            vector<ConstantSP> tArgs = {batchTable};
            batchTable = transform->call(heap, tArgs);
        }

        // Fix: To invoke the append! function, we need a vector of ConstantSP as arguments
        // appendArgs will correspond to: append!(destTable, batchTable)
        vector<ConstantSP> appendArgs = {destTable, batchTable};

        try {
            // Triggering the full DolphinDB transaction chain
            appendFunc->call(heap, appendArgs);
        } catch(exception &e) {
            duckdb_destroy_data_chunk(&chunk);
            duckdb_destroy_result(&result);
            duckdb_destroy_prepare(&stmt);
            throw RuntimeException("Append error: " + string(e.what()));
        } catch(...) {
            duckdb_destroy_data_chunk(&chunk);
            duckdb_destroy_result(&result);
            duckdb_destroy_prepare(&stmt);
            throw RuntimeException("Call append function error.");
        }

        totalInserted += accumulatedRows;


        accumulatedRows = 0;
        batchCols = createBatchCols(batchSize);
    }

    duckdb_destroy_data_chunk(&chunk);
}