* Updates a batch of updates in a single transaction. This is done optimistically * For scenarios where it's successful. Rolls back the sub-transaction in case of * failure. */
| 1120 | * failure. |
| 1121 | */ |
| 1122 | static bool |
| 1123 | DoMultiUpdate(MongoCollection *collection, List *updates, text *transactionId, |
| 1124 | BatchUpdateResult *batchResult, int updateIndex, |
| 1125 | bool forceInlineWrites, int *recordsUpdated, |
| 1126 | ExprEvalState *stateForSchemaValidation, |
| 1127 | WriteMode writeMode) |
| 1128 | { |
| 1129 | /* |
| 1130 | * Execute the query inside a sub-transaction, so we can restore order |
| 1131 | * after a failure. |
| 1132 | */ |
| 1133 | MemoryContext oldContext = CurrentMemoryContext; |
| 1134 | ResourceOwner oldOwner = CurrentResourceOwner; |
| 1135 | |
| 1136 | /* declared volatile because of the longjmp in PG_CATCH */ |
| 1137 | volatile int updateInnerIndex = updateIndex; |
| 1138 | volatile int updateCount = 0; |
| 1139 | |
| 1140 | BatchUpdateResult batchResultInner; |
| 1141 | memset(&batchResultInner, 0, sizeof(batchResultInner)); |
| 1142 | |
| 1143 | BeginInternalSubTransaction(NULL); |
| 1144 | |
| 1145 | PG_TRY(); |
| 1146 | { |
| 1147 | if (writeMode == WriteMode_Bulk_Proc) |
| 1148 | { |
| 1149 | /* Set lock timeout inside the subtransaction so it is |
| 1150 | * automatically reverted on rollback. */ |
| 1151 | char *batchLockTimeoutStr = psprintf("%d", BatchUpdateLockTimeoutMs); |
| 1152 | SetGUCLocally("lock_timeout", batchLockTimeoutStr); |
| 1153 | } |
| 1154 | ListCell *updateCell; |
| 1155 | while (updateInnerIndex < list_length(updates) && |
| 1156 | updateCount < BatchWriteSubTransactionCount) |
| 1157 | { |
| 1158 | CHECK_FOR_INTERRUPTS(); |
| 1159 | updateCell = list_nth_cell(updates, updateInnerIndex); |
| 1160 | UpdateSpec *updateSpec = lfirst(updateCell); |
| 1161 | UpdateResult updateResult = { 0 }; |
| 1162 | ProcessUpdate(collection, updateSpec, transactionId, &updateResult, |
| 1163 | forceInlineWrites, stateForSchemaValidation); |
| 1164 | UpdateResultInBatch(&batchResultInner, &updateResult, |
| 1165 | batchResult->resultMemoryContext, |
| 1166 | updateInnerIndex); |
| 1167 | updateInnerIndex++; |
| 1168 | updateCount++; |
| 1169 | } |
| 1170 | |
| 1171 | /* Commit the inner transaction, return to outer xact context */ |
| 1172 | ReleaseCurrentSubTransaction(); |
| 1173 | MemoryContextSwitchTo(oldContext); |
| 1174 | CurrentResourceOwner = oldOwner; |
| 1175 | |
| 1176 | MemoryContextSwitchTo(batchResult->resultMemoryContext); |
| 1177 | batchResult->rowsMatched += batchResultInner.rowsMatched; |
| 1178 | batchResult->rowsModified += batchResultInner.rowsModified; |
| 1179 |
no test coverage detected