LuoGe
/
openGauss-server
forked from huawei/openGauss-server
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

!179 MOT Cold start optimization and removing wrong reference in FDW state 

Merge pull request !179 from Vinoth/master
This commit is contained in:
opengauss-bot 2020-09-03 18:25:37 +08:00 committed by Gitee
parent 4df5e34cad 3b3540acc6
commit 1f17263be3
8 changed files with 159 additions and 125 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
src/gausskernel/storage/mot/core/src/storage/table.cpp
View File

@ -388,22 +388,24 @@ bool Table::CreateSecondaryIndexData(MOT::Index* index, TxnManager* txn)
return ret;
}
RC Table::InsertRowNonTransactional(Row* row, uint64_t tid, Key* k, bool isInterTest)
RC Table::InsertRowNonTransactional(Row* row, uint64_t tid, Key* k, bool skipSecIndex)
{
RC rc = RC_OK;
MaxKey m_key;
MaxKey key;
Key* pk = nullptr;
uint64_t surrogateprimaryKey = 0;
MOT::Index* ix = GetPrimaryIndex();
uint32_t numIndexes = GetNumIndexes();
SurrogateKeyGenerator& _surr_gen = GetSurrogateKeyManager()->GetSurrogateSlot(MOT_GET_CURRENT_CONNECTION_ID());
row->SetRowId(_surr_gen.GetSurrogateKey(MOT_GET_CURRENT_CONNECTION_ID()));
if (row->GetRowId() == 0) {
row->SetRowId(_surr_gen.GetSurrogateKey(MOT_GET_CURRENT_CONNECTION_ID()));
}
// add row
if (k != nullptr) {
pk = k;
} else {
pk = &m_key;
pk = &key;
pk->InitKey(ix->GetKeyLength());
// set primary key
if (ix->IsFakePrimary()) {
@ -427,18 +429,18 @@ RC Table::InsertRowNonTransactional(Row* row, uint64_t tid, Key* k, bool isInter
}
// add secondary indexes
if (!isInterTest) {
if (!skipSecIndex) {
for (uint16_t i = 1; i < numIndexes; i++) {
ix = GetSecondaryIndex(i);
m_key.InitKey(ix->GetKeyLength());
ix->BuildKey(this, row, &m_key);
if (ix->IndexInsert(&m_key, row, tid) == nullptr) {
key.InitKey(ix->GetKeyLength());
ix->BuildKey(this, row, &key);
if (ix->IndexInsert(&key, row, tid) == nullptr) {
if (MOT_IS_SEVERE()) {
MOT_REPORT_ERROR(MOT_ERROR_INTERNAL,
"Insert row",
"Failed to insert row to secondary index %u, key: %s",
i,
m_key.GetKeyStr().c_str());
key.GetKeyStr().c_str());
}
return MOT_GET_LAST_ERROR_RC();
}

4
src/gausskernel/storage/mot/core/src/storage/table.h
View File

@ -556,10 +556,10 @@ public:
* @param row. New row to be inserted
* @param tid The logical identifier of the requesting thread.
* @param k row's primary ket
* @param isInterTest determines if secondaries should be added as well
* @param skipSecIndex determines if secondaries should be added as well
* @return Status of the operation.
*/
RC InsertRowNonTransactional(Row* row, uint64_t tid, Key* k = NULL, bool isInterTest = false);
RC InsertRowNonTransactional(Row* row, uint64_t tid, Key* k = NULL, bool skipSecIndex = false);
/**
* @brief Inserts a row into a newly created secondary index storage without validation.

71
src/gausskernel/storage/mot/core/src/system/recovery/recovery_manager.cpp
View File

@ -235,11 +235,18 @@ bool RecoveryManager::RecoverTableMetadata(uint32_t tableId)
return (status == RC_OK);
}
bool RecoveryManager::RecoverTableRows(
uint32_t tableId, uint32_t seg, uint32_t tid, uint64_t& maxCsn, SurrogateState& sState)
bool RecoveryManager::RecoverTableRows(uint32_t tableId, uint32_t seg, uint32_t tid, char* keyData, char* entryData,
uint64_t& maxCsn, SurrogateState& sState)
{
RC status = RC_OK;
int fd = -1;
Table* table = nullptr;
if (!GetRecoveryManager()->FetchTable(tableId, table)) {
MOT_REPORT_ERROR(MOT_ERROR_INTERNAL, "RecoveryManager::recoverTableRows", "Table %llu does not exist", tableId);
return false;
}
std::string fileName;
CheckpointUtils::MakeCpFilename(tableId, fileName, m_workingDir, seg);
if (!CheckpointUtils::OpenFileRead(fileName, fd)) {
@ -261,22 +268,14 @@ bool RecoveryManager::RecoverTableRows(
return false;
}
uint64_t tableExId = table->GetTableExId();
if (tableExId != fileHeader.m_exId) {
MOT_LOG_ERROR(
"RecoveryManager::recoverTableRows: exId mismatch: my %lu - pkt %lu", tableExId, fileHeader.m_exId);
return false;
}
CheckpointUtils::EntryHeader entry;
char* keyData = (char*)malloc(MAX_KEY_SIZE);
if (keyData == nullptr) {
MOT_LOG_ERROR("RecoveryManager::recoverTableRows: failed to allocate key buffer");
CheckpointUtils::CloseFile(fd);
return false;
}
char* entryData = (char*)malloc(MAX_TUPLE_SIZE);
if (entryData == nullptr) {
MOT_LOG_ERROR("RecoveryManager::recoverTableRows: failed to allocate row buffer");
CheckpointUtils::CloseFile(fd);
free(keyData);
return false;
}
for (uint64_t i = 0; i < fileHeader.m_numOps; i++) {
if (IsRecoveryMemoryLimitReached(m_numWorkers)) {
MOT_LOG_ERROR("Memory hard limit reached. Cannot recover datanode");
@ -324,19 +323,16 @@ bool RecoveryManager::RecoverTableRows(
break;
}
BeginTransaction();
InsertRow(tableId,
fileHeader.m_exId,
InsertRowFromCheckpoint(table,
keyData,
entry.m_keyLen,
entryData,
entry.m_dataLen,
entry.m_csn,
tid,
m_sState,
sState,
status,
entry.m_rowId);
status = CommitTransaction(entry.m_csn);
if (status != RC_OK) {
MOT_LOG_ERROR(
"Failed to commit row recovery from checkpoint: %s (error code: %d)", RcToString(status), (int)status);
@ -354,12 +350,7 @@ bool RecoveryManager::RecoverTableRows(
seg,
fileHeader.m_numOps,
status == RC_OK ? "OK" : "Error");
if (keyData != nullptr) {
free(keyData);
}
if (entryData != nullptr) {
free(entryData);
}
return (status == RC_OK);
}
@ -384,6 +375,21 @@ void RecoveryManager::CpWorkerFunc()
"RecoveryManager::workerFunc failed to allocate surrogate state");
return;
}
char* keyData = (char*)malloc(MAX_KEY_SIZE);
if (keyData == nullptr) {
GetRecoveryManager()->OnError(
MOT::RecoveryManager::ErrCodes::CP_RECOVERY, "RecoveryManager::workerFunc: failed to allocate key buffer");
return;
}
char* entryData = (char*)malloc(MAX_TUPLE_SIZE);
if (entryData == nullptr) {
GetRecoveryManager()->OnError(
MOT::RecoveryManager::ErrCodes::CP_RECOVERY, "RecoveryManager::workerFunc: failed to allocate row buffer");
free(keyData);
return;
}
MOT_LOG_DEBUG("RecoveryManager::workerFunc start [%u] on cpu %lu", (unsigned)MOTCurrThreadId, sched_getcpu());
uint64_t maxCsn = 0;
@ -391,7 +397,7 @@ void RecoveryManager::CpWorkerFunc()
uint32_t tableId = 0;
uint32_t seg = 0;
if (GetTask(tableId, seg)) {
if (!RecoverTableRows(tableId, seg, MOTCurrThreadId, maxCsn, sState)) {
if (!RecoverTableRows(tableId, seg, MOTCurrThreadId, keyData, entryData, maxCsn, sState)) {
MOT_LOG_ERROR("RecoveryManager::workerFunc recovery of table %lu's data failed", tableId);
GetRecoveryManager()->OnError(MOT::RecoveryManager::ErrCodes::CP_RECOVERY,
"RecoveryManager::workerFunc failed to recover table: ",
@ -403,9 +409,12 @@ void RecoveryManager::CpWorkerFunc()
}
}
free(keyData);
free(entryData);
GetRecoveryManager()->SetCsnIfGreater(maxCsn);
if (sState.IsEmpty() == false) {
GetRecoveryManager()->AddSurrogateArrayToList(m_sState);
if (!sState.IsEmpty()) {
GetRecoveryManager()->AddSurrogateArrayToList(sState);
}
GetSessionManager()->DestroySessionContext(sessionContext);

23
src/gausskernel/storage/mot/core/src/system/recovery/recovery_manager.h
View File

@ -211,12 +211,15 @@ private:
* @param tableId The table id to recover.
* @param seg Segment file number to recover from.
* @param tid The current thread id
* @param keyData The key buffer to use.
* @param entryData The row buffer to use.
* @param maxCsn The returned maxCsn encountered during the recovery.
* @param sState Surrogate key state structure that will be filled
* during the recovery
* @return Boolean value denoting success or failure.
*/
bool RecoverTableRows(uint32_t tableId, uint32_t seg, uint32_t tid, uint64_t& maxCsn, SurrogateState& sState);
bool RecoverTableRows(uint32_t tableId, uint32_t seg, uint32_t tid, char* keyData, char* entryData,
uint64_t& maxCsn, SurrogateState& sState);
/**
* @brief Reads and creates a table's defenition from a checkpoint
@ -814,7 +817,7 @@ private:
/**
* @brief performs the actual row insertion to the storage.
* @param tableId the table's id.
* @param exId the the table's external id.
* @param exId the table's external id.
* @param keyData key's data buffer.
* @param keyLen key's data buffer len.
* @param rowData row's data buffer.
@ -830,6 +833,22 @@ private:
uint64_t rowLen, uint64_t csn, uint32_t tid, SurrogateState& sState, RC& status, uint64_t rowId,
bool insertLocked = false);
/**
* @brief performs non transactional row insertion (for checkpoint recovery).
* @param table the table's pointer.
* @param keyData key's data buffer.
* @param keyLen key's data buffer len.
* @param rowData row's data buffer.
* @param rowLen row's data buffer len.
* @param csn the operations's csn.
* @param tid the thread id of the recovering thread.
* @param sState the returned surrugate state.
* @param status the returned status of the operation
* @param rowId the row's internal id
*/
static void InsertRowFromCheckpoint(Table* table, char* keyData, uint16_t keyLen, char* rowData, uint64_t rowLen,
uint64_t csn, uint32_t tid, SurrogateState& sState, RC& status, uint64_t rowId);
/**
* @brief performs the actual row update in the storage.
* @param tableId the table's id.

27
src/gausskernel/storage/mot/core/src/system/recovery/recovery_ops.cpp
View File

@ -535,6 +535,33 @@ void RecoveryManager::InsertRow(uint64_t tableId, uint64_t exId, char* keyData,
}
}
void RecoveryManager::InsertRowFromCheckpoint(Table* table, char* keyData, uint16_t keyLen, char* rowData,
uint64_t rowLen, uint64_t csn, uint32_t tid, SurrogateState& sState, RC& status, uint64_t rowId)
{
MaxKey key;
Row* row = table->CreateNewRow();
if (row == nullptr) {
status = RC_ERROR;
MOT_REPORT_ERROR(MOT_ERROR_OOM, "Recovery Manager Insert Row", "failed to create row");
return;
}
row->CopyData((const uint8_t*)rowData, rowLen);
row->SetCommitSequenceNumber(csn);
row->SetRowId(rowId);
MOT::Index* ix = table->GetPrimaryIndex();
if (ix->IsFakePrimary()) {
row->SetSurrogateKey(*(uint64_t*)keyData);
sState.UpdateMaxKey(rowId);
}
key.CpKey((const uint8_t*)keyData, keyLen);
status = table->InsertRowNonTransactional(row, tid, &key);
if (status != RC_OK) {
MOT_REPORT_ERROR(MOT_ERROR_OOM, "Recovery Manager Insert Row", "failed to insert row");
table->DestroyRow(row);
}
}
void RecoveryManager::DeleteRow(
uint64_t tableId, uint64_t exId, char* keyData, uint16_t keyLen, uint64_t csn, uint32_t tid, RC& status)
{

130
src/gausskernel/storage/mot/fdw_adapter/src/mot_fdw.cpp
View File

@ -1002,13 +1002,13 @@ static TupleTableSlot* MOTIterateForeignScan(ForeignScanState* node)
return nullptr;
}
MOT::Row* currRow;
MOTFdwStateSt* festate = (MOTFdwStateSt*)node->fdw_state;
TupleTableSlot* slot = node->ss.ss_ScanTupleSlot;
bool found = false;
bool stopAtFirst = (festate->m_bestIx && festate->m_bestIx->m_ixOpers[0] == KEY_OPER::READ_KEY_EXACT &&
festate->m_bestIx->m_ix->GetUnique() == true);
festate->m_currRow = NULL;
(void)ExecClearTuple(slot);
if (stopAtFirst) {
@ -1018,18 +1018,18 @@ static TupleTableSlot* MOTIterateForeignScan(ForeignScanState* node)
MOTAdaptor::CreateKeyBuffer(node->ss.ss_currentRelation, festate, 0);
MOT::Sentinel* Sentinel =
festate->m_bestIx->m_ix->IndexReadSentinel(&festate->m_stateKey[0], festate->m_currTxn->GetThdId());
festate->m_currRow = festate->m_currTxn->RowLookup(festate->m_internalCmdOper, Sentinel, rc);
currRow = festate->m_currTxn->RowLookup(festate->m_internalCmdOper, Sentinel, rc);
if (festate->m_currRow != NULL) {
if (currRow != NULL) {
MOTAdaptor::UnpackRow(
slot, festate->m_table, festate->m_attrsUsed, const_cast<uint8_t*>(festate->m_currRow->GetData()));
slot, festate->m_table, festate->m_attrsUsed, const_cast<uint8_t*>(currRow->GetData()));
node->ss.is_scan_end = true;
fscan->scan.scan_qual_optimized = true;
ExecStoreVirtualTuple(slot);
if (festate->m_ctidNum > 0) {
HeapTuple resultTup = ExecFetchSlotTuple(slot);
MOTRecConvertSt cv;
cv.m_u.m_ptr = (uint64_t)festate->m_currRow->GetPrimarySentinel();
cv.m_u.m_ptr = (uint64_t)currRow->GetPrimarySentinel();
resultTup->t_self = cv.m_u.m_self;
HeapTupleSetXmin(resultTup, InvalidTransactionId);
HeapTupleSetXmax(resultTup, InvalidTransactionId);
@ -1083,10 +1083,8 @@ static TupleTableSlot* MOTIterateForeignScan(ForeignScanState* node)
do {
MOT::Sentinel* Sentinel = festate->m_cursor[0]->GetPrimarySentinel();
festate->m_currRow = festate->m_currTxn->RowLookup(festate->m_internalCmdOper, Sentinel, rc);
if (festate->m_currRow == NULL) {
currRow = festate->m_currTxn->RowLookup(festate->m_internalCmdOper, Sentinel, rc);
if (currRow == NULL) {
if (rc != MOT::RC_OK) {
if (MOT_IS_SEVERE()) {
MOT_REPORT_ERROR(MOT_ERROR_INTERNAL, "MOTIterateForeignScan", "Failed to lookup row");
@ -1113,7 +1111,7 @@ static TupleTableSlot* MOTIterateForeignScan(ForeignScanState* node)
}
MOTAdaptor::UnpackRow(
slot, festate->m_table, festate->m_attrsUsed, const_cast<uint8_t*>(festate->m_currRow->GetData()));
slot, festate->m_table, festate->m_attrsUsed, const_cast<uint8_t*>(currRow->GetData()));
found = true;
festate->m_cursor[0]->Next();
@ -1126,7 +1124,7 @@ static TupleTableSlot* MOTIterateForeignScan(ForeignScanState* node)
if (festate->m_ctidNum > 0) {
HeapTuple resultTup = ExecFetchSlotTuple(slot);
MOTRecConvertSt cv;
cv.m_u.m_ptr = (uint64_t)festate->m_currRow->GetPrimarySentinel();
cv.m_u.m_ptr = (uint64_t)currRow->GetPrimarySentinel();
resultTup->t_self = cv.m_u.m_self;
HeapTupleSetXmin(resultTup, InvalidTransactionId);
HeapTupleSetXmax(resultTup, InvalidTransactionId);
@ -1456,58 +1454,48 @@ static TupleTableSlot* MOTExecForeignUpdate(
{
MOTFdwStateSt* fdwState = (MOTFdwStateSt*)resultRelInfo->ri_FdwState;
MOT::RC rc = MOT::RC_OK;
TupleTableSlot* dataSlot = slot;
bool cleanCurrRow = false;
MOT::Row* currRow;
AttrNumber num = fdwState->m_ctidNum - 1;
MOTRecConvertSt cv;
if (MOTAdaptor::m_engine->IsSoftMemoryLimitReached()) {
CleanQueryStatesOnError(fdwState->m_currTxn);
}
isMemoryLimitReached();
if (fdwState->m_currRow == nullptr) {
AttrNumber num = fdwState->m_ctidNum - 1;
if (fdwState->m_ctidNum != 0 && planSlot->tts_nvalid >= fdwState->m_ctidNum && !planSlot->tts_isnull[num]) {
MOTRecConvertSt cv;
cv.m_u.m_ptr = 0;
cv.m_u.m_self = *(ItemPointerData*)planSlot->tts_values[num];
fdwState->m_currRow =
fdwState->m_currTxn->RowLookup(fdwState->m_internalCmdOper, (MOT::Sentinel*)cv.m_u.m_ptr, rc);
}
if (fdwState->m_currRow == nullptr) {
CleanQueryStatesOnError(fdwState->m_currTxn);
report_pg_error(MOT::RC_ERROR, fdwState->m_currTxn);
return nullptr;
}
cleanCurrRow = true;
if (slot->tts_nvalid == 0)
dataSlot = planSlot;
if (fdwState->m_ctidNum != 0 && planSlot->tts_nvalid >= fdwState->m_ctidNum && !planSlot->tts_isnull[num]) {
cv.m_u.m_ptr = 0;
cv.m_u.m_self = *(ItemPointerData*)planSlot->tts_values[num];
currRow = fdwState->m_currTxn->RowLookup(fdwState->m_internalCmdOper, (MOT::Sentinel*)cv.m_u.m_ptr, rc);
} else {
fdwState->m_currRow =
fdwState->m_currTxn->RowLookup(fdwState->m_internalCmdOper, fdwState->m_currRow->GetPrimarySentinel(), rc);
elog(ERROR, "MOTExecForeignUpdate failed to fetch row for update ctid %d nvalid %d %s",
num, planSlot->tts_nvalid, (planSlot->tts_isnull[num] ? "NULL" : "NOT NULL"));
CleanQueryStatesOnError(fdwState->m_currTxn);
report_pg_error(MOT::RC_ERROR, fdwState->m_currTxn);
return nullptr;
}
if ((rc = MOTAdaptor::UpdateRow(fdwState, dataSlot)) == MOT::RC_OK) {
if (cleanCurrRow)
fdwState->m_currRow = nullptr;
if (currRow == nullptr) {
elog(ERROR, "MOTExecForeignUpdate failed to fetch row");
CleanQueryStatesOnError(fdwState->m_currTxn);
report_pg_error(((rc == MOT::RC_OK) ? MOT::RC_ERROR : rc), fdwState->m_currTxn);
return nullptr;
}
if (resultRelInfo->ri_projectReturning)
return dataSlot;
else {
if ((rc = MOTAdaptor::UpdateRow(fdwState, planSlot, currRow)) == MOT::RC_OK) {
if (resultRelInfo->ri_projectReturning) {
return planSlot;
} else {
estate->es_processed++;
return nullptr;
}
} else {
if (cleanCurrRow)
fdwState->m_currRow = nullptr;
elog(DEBUG2, "Abort parent transaction from MOT update, id %lu", fdwState->m_txnId);
CleanQueryStatesOnError(fdwState->m_currTxn);
report_pg_error(rc,
fdwState->m_currTxn,
(void*)(fdwState->m_currTxn->m_errIx != NULL ? fdwState->m_currTxn->m_errIx->GetName().c_str() : "unknown"),
(void*)(fdwState->m_currTxn->m_errIx != nullptr ? fdwState->m_currTxn->m_errIx->GetName().c_str()
: "unknown"),
(void*)fdwState->m_currTxn->m_errMsgBuf);
return nullptr;
}
@ -1518,50 +1506,40 @@ static TupleTableSlot* MOTExecForeignDelete(
{
MOTFdwStateSt* fdwState = (MOTFdwStateSt*)resultRelInfo->ri_FdwState;
MOT::RC rc = MOT::RC_OK;
bool cleanCurrRow = false;
MOT::Row* currRow;
AttrNumber num = fdwState->m_ctidNum - 1;
MOTRecConvertSt cv;
if (fdwState->m_currRow == nullptr) {
AttrNumber num = fdwState->m_ctidNum - 1;
if (fdwState->m_ctidNum != 0 && planSlot->tts_nvalid >= fdwState->m_ctidNum && !planSlot->tts_isnull[num]) {
MOTRecConvertSt cv;
cv.m_u.m_ptr = 0;
cv.m_u.m_self = *(ItemPointerData*)planSlot->tts_values[num];
fdwState->m_currRow =
fdwState->m_currTxn->RowLookup(fdwState->m_internalCmdOper, (MOT::Sentinel*)cv.m_u.m_ptr, rc);
}
if (fdwState->m_currRow == nullptr) {
CleanQueryStatesOnError(fdwState->m_currTxn);
report_pg_error(MOT::RC_ERROR, fdwState->m_currTxn);
return nullptr;
}
cleanCurrRow = true;
if (fdwState->m_ctidNum != 0 && planSlot->tts_nvalid >= fdwState->m_ctidNum && !planSlot->tts_isnull[num]) {
cv.m_u.m_ptr = 0;
cv.m_u.m_self = *(ItemPointerData*)planSlot->tts_values[num];
currRow = fdwState->m_currTxn->RowLookup(fdwState->m_internalCmdOper, (MOT::Sentinel*)cv.m_u.m_ptr, rc);
} else {
fdwState->m_currRow =
fdwState->m_currTxn->RowLookup(fdwState->m_internalCmdOper, fdwState->m_currRow->GetPrimarySentinel(), rc);
elog(ERROR, "MOTExecForeignDelete failed to fetch row for delete ctid %d nvalid %d %s",
num, planSlot->tts_nvalid, (planSlot->tts_isnull[num] ? "NULL" : "NOT NULL"));
CleanQueryStatesOnError(fdwState->m_currTxn);
report_pg_error(MOT::RC_ERROR, fdwState->m_currTxn);
return nullptr;
}
if (currRow == nullptr) {
elog(ERROR, "MOTExecForeignDelete failed to fetch row");
CleanQueryStatesOnError(fdwState->m_currTxn);
report_pg_error(((rc == MOT::RC_OK) ? MOT::RC_ERROR : rc), fdwState->m_currTxn);
return nullptr;
}
if ((rc = MOTAdaptor::DeleteRow(fdwState, slot)) == MOT::RC_OK) {
if (resultRelInfo->ri_projectReturning) {
MOTAdaptor::UnpackRow(
slot, fdwState->m_table, fdwState->m_attrsUsed, const_cast<uint8_t*>(fdwState->m_currRow->GetData()));
slot, fdwState->m_table, fdwState->m_attrsUsed, const_cast<uint8_t*>(currRow->GetData()));
ExecStoreVirtualTuple(slot);
if (cleanCurrRow)
fdwState->m_currRow = nullptr;
return slot;
} else {
if (cleanCurrRow)
fdwState->m_currRow = nullptr;
estate->es_processed++;
return nullptr;
}
} else {
if (cleanCurrRow)
fdwState->m_currRow = nullptr;
elog(DEBUG2, "Abort parent transaction from MOT delete, id %lu", fdwState->m_txnId);
CleanQueryStatesOnError(fdwState->m_currTxn);
report_pg_error(rc,

6
src/gausskernel/storage/mot/fdw_adapter/src/mot_internal.cpp
View File

@ -1138,7 +1138,7 @@ MOT::RC MOTAdaptor::InsertRow(MOTFdwStateSt* fdwState, TupleTableSlot* slot)
return res;
}
MOT::RC MOTAdaptor::UpdateRow(MOTFdwStateSt* fdwState, TupleTableSlot* slot)
MOT::RC MOTAdaptor::UpdateRow(MOTFdwStateSt* fdwState, TupleTableSlot* slot, MOT::Row* currRow)
{
EnsureSafeThreadAccessInline();
MOT::RC rc;
@ -1149,11 +1149,11 @@ MOT::RC MOTAdaptor::UpdateRow(MOTFdwStateSt* fdwState, TupleTableSlot* slot)
if (rc != MOT::RC::RC_OK) {
break;
}
uint8_t* rowData = const_cast<uint8_t*>(fdwState->m_currRow->GetData());
uint8_t* rowData = const_cast<uint8_t*>(currRow->GetData());
PackUpdateRow(slot, fdwState->m_table, fdwState->m_attrsModified, rowData);
MOT::BitmapSet modified_columns(fdwState->m_attrsModified, fdwState->m_table->GetFieldCount() - 1);
rc = fdwState->m_currTxn->OverwriteRow(fdwState->m_currRow, modified_columns);
rc = fdwState->m_currTxn->OverwriteRow(currRow, modified_columns);
} while (0);
return rc;

3
src/gausskernel/storage/mot/fdw_adapter/src/mot_internal.h
View File

@ -273,7 +273,6 @@ struct MOTFdwState_St {
MOT::Table* m_table;
MOT::IndexIterator* m_cursor[2] = {nullptr, nullptr};
MOT::TxnManager* m_currTxn;
MOT::Row* m_currRow = nullptr;
void* m_currItem = nullptr;
uint32_t m_rowsFound = 0;
bool m_cursorOpened = false;
@ -338,7 +337,7 @@ public:
static MOT::RC RollbackPrepared(TransactionId tid);
static MOT::RC FailedCommitPrepared(TransactionId tid);
static MOT::RC InsertRow(MOTFdwStateSt* fdwState, TupleTableSlot* slot);
static MOT::RC UpdateRow(MOTFdwStateSt* fdwState, TupleTableSlot* slot);
static MOT::RC UpdateRow(MOTFdwStateSt* fdwState, TupleTableSlot* slot, MOT::Row* currRow);
static MOT::RC DeleteRow(MOTFdwStateSt* fdwState, TupleTableSlot* slot);
/* Convertors */