* UpdateOneObjectId handles the case where we are updating a single document * by _id from a collection that is sharded on some other key. In this case, * we need to look across all shards for a matching _id, then update only that * one. * * Citus does not support SELECT .. FOR UPDATE, and it is very difficult to * support efficiently without running into frequent deadlocks. Therefore, * we
| 3818 | * be deleted or updated concurrently. In that case, we try again. |
| 3819 | */ |
| 3820 | static void |
| 3821 | UpdateOneObjectId(MongoCollection *collection, UpdateOneParams *updateOneParams, |
| 3822 | bson_value_t *objectId, bool queryHasNonIdFilters, text *transactionId, |
| 3823 | UpdateOneResult *result, ExprEvalState *stateForSchemaValidation) |
| 3824 | { |
| 3825 | /* initialize result */ |
| 3826 | memset(result, 0, sizeof(UpdateOneResult)); |
| 3827 | |
| 3828 | const int maxTries = 5; |
| 3829 | |
| 3830 | if (transactionId != NULL) |
| 3831 | { |
| 3832 | RetryableWriteResult writeResult; |
| 3833 | |
| 3834 | /* |
| 3835 | * Try to find a retryable write record for the transaction ID in any shard. |
| 3836 | */ |
| 3837 | if (FindRetryRecordInAnyShard(collection->collectionId, transactionId, |
| 3838 | &writeResult)) |
| 3839 | { |
| 3840 | /* found a retry record, return the previous result */ |
| 3841 | result->isRetry = true; |
| 3842 | result->isRowUpdated = writeResult.rowsAffected; |
| 3843 | |
| 3844 | /* these writes are never upserts */ |
| 3845 | result->upsertedObjectId = NULL; |
| 3846 | |
| 3847 | return; |
| 3848 | } |
| 3849 | } |
| 3850 | |
| 3851 | for (int tryNumber = 0; tryNumber < maxTries; tryNumber++) |
| 3852 | { |
| 3853 | int64 shardKeyValue = 0; |
| 3854 | bool isIdValueCollationAware = false; |
| 3855 | const char *collationStringIgnore = NULL; |
| 3856 | if (!FindShardKeyValueForDocumentId(collection, updateOneParams->query, objectId, |
| 3857 | isIdValueCollationAware, queryHasNonIdFilters, |
| 3858 | &shardKeyValue, updateOneParams->variableSpec, |
| 3859 | collationStringIgnore)) |
| 3860 | { |
| 3861 | /* no document matches both the query and the object ID */ |
| 3862 | return; |
| 3863 | } |
| 3864 | |
| 3865 | /* we do not support upsert without shard key filter */ |
| 3866 | Assert(updateOneParams->isUpsert == false); |
| 3867 | |
| 3868 | bool forceInlineWrites = false; |
| 3869 | CallUpdateOne(collection, updateOneParams, shardKeyValue, |
| 3870 | transactionId, result, forceInlineWrites, stateForSchemaValidation); |
| 3871 | |
| 3872 | if (result->isRowUpdated || result->updateSkipped) |
| 3873 | { |
| 3874 | if (result->reinsertDocument != NULL) |
| 3875 | { |
| 3876 | /* we could support reinsert here? */ |
| 3877 | ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), |
no test coverage detected