diff --git a/sdk/cosmos/.cspell.json b/sdk/cosmos/.cspell.json index 3cd11f77289..117ff7d1c0e 100644 --- a/sdk/cosmos/.cspell.json +++ b/sdk/cosmos/.cspell.json @@ -13,6 +13,7 @@ "autoscale", "azurecosmos", "backoff", + "backpressure", "bytearray", "canadacentral", "canadaeast", @@ -26,9 +27,11 @@ "cloneable", "colls", "cosmosclient", + "cutover", "Daad", "dedicatedgateway", "derefs", + "dhat", "Dmaster", "documentdb", "dotproduct", @@ -40,10 +43,12 @@ "enablepreview", "EUAP", "euclidian", + "evals", "fabianm", "failback", "failovers", "FILETIME", + "flamegraph", "fract", "francecentral", "francesouth", @@ -54,6 +59,7 @@ "germanywestcentral", "hostnames", "hotfixes", + "idents", "IMDS", "inclusivity", "isquery", @@ -62,7 +68,10 @@ "keepalive", "koreacentral", "koreasouth", + "Kusto", "linearizability", + "LLZA", + "memcheck", "MEMORYSTATUSEX", "moka", "multihash", @@ -86,6 +95,11 @@ "othercoll", "partitionkey", "partitionkeyrangeid", + "perfcontainer", + "perfdb", + "perfresults", + "PIDS", + "Pigw", "Pkrange", "pkranges", "pksysdocs", @@ -96,17 +110,20 @@ "PPAF", "PPCB", "pushback", + "pyroscope", "qname", + "qself", "RAII", "readfeed", "replicaset", - "reqs", "Replicaset", + "reqs", "Retriable", "retryable", "rfind", "RNTBD", "roundtrips", + "RUPM", "rwcache", "sess", "southafricanorth", @@ -179,4 +196,4 @@ ] } ] -} +} \ No newline at end of file diff --git a/sdk/cosmos/azure_data_cosmos/CHANGELOG.md b/sdk/cosmos/azure_data_cosmos/CHANGELOG.md index 6df0aaf05da..2896cf54fc9 100644 --- a/sdk/cosmos/azure_data_cosmos/CHANGELOG.md +++ b/sdk/cosmos/azure_data_cosmos/CHANGELOG.md @@ -15,6 +15,7 @@ ### Breaking Changes +- `ContainerClient::create_item()` and `ContainerClient::upsert_item()` now require an `item_id: &str` parameter (same pattern as `replace_item` and `read_item`). The item id is passed to the driver via `ItemReference` so the body never needs to be parsed to extract the document id. - Renamed `replace_throughput` to `begin_replace_throughput` on `ContainerClient` and `DatabaseClient`. The return type changed from `ResourceResponse` to `ThroughputPoller`. ([#4096](https://github.com/Azure/azure-sdk-for-rust/pull/4096)) - Removed `CreateDatabaseOptions::with_throughput()`. Database-level shared throughput provisioning is no longer supported through the SDK. Use container-level throughput instead. ([#4147](https://github.com/Azure/azure-sdk-for-rust/pull/4147)) diff --git a/sdk/cosmos/azure_data_cosmos/README.md b/sdk/cosmos/azure_data_cosmos/README.md index c5c7c4066e7..8d68b68acd5 100644 --- a/sdk/cosmos/azure_data_cosmos/README.md +++ b/sdk/cosmos/azure_data_cosmos/README.md @@ -108,7 +108,7 @@ async fn example(cosmos_client: CosmosClient) -> Result<(), Box { @@ -94,7 +99,9 @@ impl CreateCommand { None }; - let response = container_client.create_item(pk, item, options).await?; + let response = container_client + .create_item(pk, &item_id, item, options) + .await?; println!("Created item successfully"); diff --git a/sdk/cosmos/azure_data_cosmos/examples/cosmos/upsert.rs b/sdk/cosmos/azure_data_cosmos/examples/cosmos/upsert.rs index 4e77d466d97..dc033906874 100644 --- a/sdk/cosmos/azure_data_cosmos/examples/cosmos/upsert.rs +++ b/sdk/cosmos/azure_data_cosmos/examples/cosmos/upsert.rs @@ -21,6 +21,10 @@ pub struct UpsertCommand { #[arg(long, short)] partition_key: String, + /// The id of the new item. + #[arg(long, short)] + item_id: String, + /// The JSON of the new item. #[arg(long, short)] json: String, @@ -46,7 +50,9 @@ impl UpsertCommand { None }; - let response = container_client.upsert_item(pk, item, options).await?; + let response = container_client + .upsert_item(pk, &self.item_id, item, options) + .await?; println!("Item updated successfully"); if self.show_updated { diff --git a/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs b/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs index fc6fb790ea9..f3fdf385f4f 100644 --- a/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs +++ b/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs @@ -254,6 +254,7 @@ impl ContainerClient { /// /// # Arguments /// * `partition_key` - The partition key of the new item. + /// * `item_id` - The id of the new item. /// * `item` - The item to create. The type must implement [`Serialize`] and [`Deserialize`](serde::Deserialize) /// * `options` - Optional parameters for the request /// @@ -276,7 +277,7 @@ impl ContainerClient { /// }; /// # let container_client: azure_data_cosmos::clients::ContainerClient = panic!("this is a non-running example"); /// container_client - /// .create_item("category1", p, None) + /// .create_item("category1", "product1", p, None) /// .await?; /// # } /// ``` @@ -308,7 +309,7 @@ impl ContainerClient { /// operation.content_response_on_write = Some(ContentResponseOnWrite::Enabled); /// let options = ItemWriteOptions::default().with_operation_options(operation); /// let created_item = container_client - /// .create_item("category1", p, Some(options)) + /// .create_item("category1", "product1", p, Some(options)) /// .await? /// .into_body().json::(); /// # Ok(()) @@ -317,16 +318,22 @@ impl ContainerClient { pub async fn create_item( &self, partition_key: impl Into, + item_id: &str, item: T, options: Option, ) -> azure_core::Result> { let options = options.unwrap_or_default(); let body = serde_json::to_vec(&item)?; - let driver_pk = partition_key.into().into_driver_partition_key(); + + // Build the driver's item reference from our stored container metadata. + let item_ref = ItemReference::from_name( + &self.container_ref, + partition_key.into().into_driver_partition_key(), + item_id.to_owned(), + ); // Create the driver operation and apply ItemWriteOptions fields. - let operation = - CosmosOperation::create_item(self.container_ref.clone(), driver_pk).with_body(body); + let operation = CosmosOperation::create_item(item_ref).with_body(body); let operation = apply_item_options(operation, options.session_token, options.precondition); // Execute through the driver. @@ -447,6 +454,7 @@ impl ContainerClient { /// /// # Arguments /// * `partition_key` - The partition key of the item to create or replace. + /// * `item_id` - The id of the item to create or replace. /// * `item` - The item to create. The type must implement [`Serialize`] and [`Deserialize`](serde::Deserialize) /// * `options` - Optional parameters for the request /// @@ -469,7 +477,7 @@ impl ContainerClient { /// }; /// # let container_client: azure_data_cosmos::clients::ContainerClient = panic!("this is a non-running example"); /// container_client - /// .upsert_item("category1", p, None) + /// .upsert_item("category1", "product1", p, None) /// .await?; /// # Ok(()) /// # } @@ -502,7 +510,7 @@ impl ContainerClient { /// operation.content_response_on_write = Some(ContentResponseOnWrite::Enabled); /// let options = ItemWriteOptions::default().with_operation_options(operation); /// let updated_product = container_client - /// .upsert_item("category1", p, Some(options)) + /// .upsert_item("category1", "product1", p, Some(options)) /// .await? /// .into_body().json::()?; /// Ok(()) @@ -510,16 +518,22 @@ impl ContainerClient { pub async fn upsert_item( &self, partition_key: impl Into, + item_id: &str, item: T, options: Option, ) -> azure_core::Result> { let options = options.unwrap_or_default(); let body = serde_json::to_vec(&item)?; - let driver_pk = partition_key.into().into_driver_partition_key(); + + // Build the driver's item reference from our stored container metadata. + let item_ref = ItemReference::from_name( + &self.container_ref, + partition_key.into().into_driver_partition_key(), + item_id.to_owned(), + ); // Create the driver operation and apply ItemWriteOptions fields. - let operation = - CosmosOperation::upsert_item(self.container_ref.clone(), driver_pk).with_body(body); + let operation = CosmosOperation::upsert_item(item_ref).with_body(body); let operation = apply_item_options(operation, options.session_token, options.precondition); // Execute through the driver. @@ -954,9 +968,9 @@ mod tests { assert_send(client.read_throughput(todo!())); assert_send(client.begin_replace_throughput(todo!(), todo!())); assert_send(client.delete(todo!())); - assert_send(client.create_item::("", todo!(), todo!())); + assert_send(client.create_item::("", todo!(), todo!(), todo!())); assert_send(client.replace_item::("", todo!(), todo!(), todo!())); - assert_send(client.upsert_item::("", todo!(), todo!())); + assert_send(client.upsert_item::("", todo!(), todo!(), todo!())); assert_send(client.read_item::("", todo!(), todo!())); assert_send(client.delete_item("", todo!(), todo!())); assert_send(client.execute_transactional_batch(todo!(), todo!())); diff --git a/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_batch.rs b/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_batch.rs index 44aa4c91967..79b61e116a0 100644 --- a/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_batch.rs +++ b/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_batch.rs @@ -132,10 +132,10 @@ pub async fn batch_mixed_operations() -> Result<(), Box> { }; container_client - .create_item(&partition_key, &item1, None) + .create_item(&partition_key, &item1.id, &item1, None) .await?; container_client - .create_item(&partition_key, &item2, None) + .create_item(&partition_key, &item2.id, &item2, None) .await?; // Now execute a batch with mixed operations @@ -206,7 +206,7 @@ pub async fn batch_atomicity_on_failure() -> Result<(), Box> { }; container_client - .create_item(&partition_key, &item1, None) + .create_item(&partition_key, &item1.id, &item1, None) .await?; // Try to create a batch that will fail (trying to delete a non-existent item) diff --git a/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_fault_injection.rs b/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_fault_injection.rs index ebb2908a68c..75213af2319 100644 --- a/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_fault_injection.rs +++ b/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_fault_injection.rs @@ -87,7 +87,9 @@ pub async fn fault_injection_probability_zero_never_fails() -> Result<(), Box Result<(), Box Result<(), Box Result<(), Box let item_id = format!("Item-{}", unique_id); // Create using normal client - container_client.create_item(&pk, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() @@ -315,7 +323,7 @@ pub async fn fault_injection_delete_item_fault_crud_succeeds() -> Result<(), Box let mut updated_item = item.clone(); updated_item.value = 100; let upsert_result = fault_container_client - .upsert_item(&pk, &updated_item, None) + .upsert_item(&pk, &item_id, &updated_item, None) .await; assert!( upsert_result.is_ok(), @@ -381,7 +389,9 @@ pub async fn fault_injection_container_specific() -> Result<(), Box> let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() @@ -482,7 +492,9 @@ pub async fn fault_injection_multiple_rules_priority() -> Result<(), Box Result<( let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() @@ -638,7 +652,9 @@ pub async fn fault_injection_first_rule_expired_due_to_end_time() -> Result<(), let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() @@ -709,7 +725,9 @@ pub async fn fault_injection_hit_limit_behavior() -> Result<(), Box> let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() @@ -775,7 +793,9 @@ pub async fn fault_injection_empty_rules() -> Result<(), Box> { let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() @@ -871,7 +891,9 @@ pub async fn fault_injection_metadata_fault_item_ops_succeed() -> Result<(), Box let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - let create_result = fault_container_client.create_item(&pk, &item, None).await; + let create_result = fault_container_client + .create_item(&pk, &item_id, &item, None) + .await; assert!( create_result.is_ok(), "create item should succeed: {:?}", @@ -892,7 +914,7 @@ pub async fn fault_injection_metadata_fault_item_ops_succeed() -> Result<(), Box let mut updated_item = item.clone(); updated_item.value = 999; let upsert_result = fault_container_client - .upsert_item(&pk, &updated_item, None) + .upsert_item(&pk, &item_id, &updated_item, None) .await; assert!( upsert_result.is_ok(), @@ -952,7 +974,9 @@ pub async fn fault_injection_enable_disable_rule() -> Result<(), Box> let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() diff --git a/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_items.rs b/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_items.rs index 2bad01200be..528a156363e 100644 --- a/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_items.rs +++ b/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_items.rs @@ -135,7 +135,9 @@ pub async fn item_crud() -> Result<(), Box> { let pk = format!("Partition@1-{}", unique_id); let item_id = format!("Item@1-{}", unique_id); - let response = container_client.create_item(&pk, &item, None).await?; + let response = container_client + .create_item(&pk, &item_id, &item, None) + .await?; assert_response( &response, StatusCode::Created, @@ -257,7 +259,9 @@ pub async fn item_read_system_properties() -> Result<(), Box> { let pk = format!("Partition1-{}", unique_id); let item_id = format!("Item1-{}", unique_id); - let create_response = container_client.create_item(&pk, &item, None).await?; + let create_response = container_client + .create_item(&pk, &item_id, &item, None) + .await?; assert_response( &create_response, StatusCode::Created, @@ -316,7 +320,9 @@ pub async fn item_upsert_new() -> Result<(), Box> { let pk = format!("Partition1-{}", unique_id); let item_id = format!("Item1-{}", unique_id); - let upsert_response = container_client.upsert_item(&pk, &item, None).await?; + let upsert_response = container_client + .upsert_item(&pk, &item_id, &item, None) + .await?; assert_response( &upsert_response, StatusCode::Created, @@ -365,8 +371,11 @@ pub async fn item_upsert_existing() -> Result<(), Box> { }; let pk = format!("Partition1-{}", unique_id); + let item_id = format!("Item1-{}", unique_id); - let create_response = container_client.create_item(&pk, &item, None).await?; + let create_response = container_client + .create_item(&pk, &item_id, &item, None) + .await?; assert_response( &create_response, StatusCode::Created, @@ -378,7 +387,7 @@ pub async fn item_upsert_existing() -> Result<(), Box> { item.nested.nested_value = "Updated".into(); let upsert_response = container_client - .upsert_item(&pk, &item, { + .upsert_item(&pk, &item_id, &item, { let mut operation = OperationOptions::default(); operation.content_response_on_write = Some(ContentResponseOnWrite::Enabled); Some(ItemWriteOptions::default().with_operation_options(operation)) @@ -424,7 +433,7 @@ pub async fn item_null_partition_key() -> Result<(), Box> { let item_id = format!("Item1-{}", unique_id); let create_response = container_client - .create_item(PartitionKey::NULL, &item, None) + .create_item(PartitionKey::NULL, &item_id, &item, None) .await?; assert_response( &create_response, @@ -437,7 +446,7 @@ pub async fn item_null_partition_key() -> Result<(), Box> { item.nested.nested_value = "Updated".into(); let upsert_response = container_client - .upsert_item(PartitionKey::NULL, &item, None) + .upsert_item(PartitionKey::NULL, &item_id, &item, None) .await?; assert_response( &upsert_response, @@ -520,7 +529,9 @@ pub async fn item_replace_if_match_etag() -> Result<(), Box> { let pk = format!("Partition1-{}", unique_id); let item_id = format!("Item1-{}", unique_id); - let response = container_client.create_item(&pk, &item, None).await?; + let response = container_client + .create_item(&pk, &item_id, &item, None) + .await?; assert_response( &response, StatusCode::Created, @@ -610,8 +621,11 @@ pub async fn item_upsert_if_match_etag() -> Result<(), Box> { }; let pk = format!("Partition1-{}", unique_id); + let item_id = format!("Item1-{}", unique_id); - let response = container_client.create_item(&pk, &item, None).await?; + let response = container_client + .create_item(&pk, &item_id, &item, None) + .await?; assert_response( &response, StatusCode::Created, @@ -633,6 +647,7 @@ pub async fn item_upsert_if_match_etag() -> Result<(), Box> { let upsert_response = container_client .upsert_item( &pk, + &item_id, &item, Some( ItemWriteOptions::default() @@ -654,6 +669,7 @@ pub async fn item_upsert_if_match_etag() -> Result<(), Box> { let response = container_client .upsert_item( &pk, + &item_id, &item, Some( ItemWriteOptions::default() @@ -701,7 +717,9 @@ pub async fn item_delete_if_match_etag() -> Result<(), Box> { let pk = format!("Partition1-{}", unique_id); let item_id = format!("Item1-{}", unique_id); - let response = container_client.create_item(&pk, &item, None).await?; + let response = container_client + .create_item(&pk, &item_id, &item, None) + .await?; assert_response( &response, StatusCode::Created, @@ -735,7 +753,9 @@ pub async fn item_delete_if_match_etag() -> Result<(), Box> { ); //Add item again for second delete test - let create_response = container_client.create_item(&pk, &item, None).await?; + let create_response = container_client + .create_item(&pk, &item_id, &item, None) + .await?; assert_response( &create_response, StatusCode::Created, @@ -803,7 +823,7 @@ pub async fn item_undefined_partition_key() -> Result<(), Box> { let item_no_pk_id = format!("Item-NoPK-{}", unique_id); let response = container_client - .create_item(PartitionKey::UNDEFINED, &item_no_pk, None) + .create_item(PartitionKey::UNDEFINED, &item_no_pk_id, &item_no_pk, None) .await?; assert_response( &response, @@ -825,7 +845,7 @@ pub async fn item_undefined_partition_key() -> Result<(), Box> { let item_null_pk_id = format!("Item-NullPK-{}", unique_id); let response = container_client - .create_item(PartitionKey::NULL, &item_null_pk, None) + .create_item(PartitionKey::NULL, &item_null_pk_id, &item_null_pk, None) .await?; assert_response( &response, @@ -844,7 +864,7 @@ pub async fn item_undefined_partition_key() -> Result<(), Box> { let item_with_pk_id = format!("Item-WithPK-{}", unique_id); let response = container_client - .create_item(&pk_value, &item_with_pk, None) + .create_item(&pk_value, &item_with_pk_id, &item_with_pk, None) .await?; assert_response( &response, @@ -955,9 +975,12 @@ pub async fn create_item_duplicate_returns_conflict() -> Result<(), Box Result<(), Box Result<(), Box> { bool_value: false, }; let pk = format!("pk-{}", unique_id); + let item_id = format!("cr-{}", unique_id); let mut operation = OperationOptions::default(); operation.content_response_on_write = Some(ContentResponseOnWrite::Enabled); let options = ItemWriteOptions::default().with_operation_options(operation); let response = container_client - .create_item(&pk, &item, Some(options)) + .create_item(&pk, &item_id, &item, Some(options)) .await?; assert_response( &response, @@ -1053,8 +1079,11 @@ pub async fn create_item_response_metadata() -> Result<(), Box> { bool_value: true, }; let pk = format!("pk-{}", unique_id); + let item_id = format!("meta-{}", unique_id); - let response = container_client.create_item(&pk, &item, None).await?; + let response = container_client + .create_item(&pk, &item_id, &item, None) + .await?; assert_eq!(response.status(), StatusCode::Created); // Session token must be present for session consistency. diff --git a/sdk/cosmos/azure_data_cosmos/tests/framework/test_data.rs b/sdk/cosmos/azure_data_cosmos/tests/framework/test_data.rs index 1a47368f9e6..99ec7849347 100644 --- a/sdk/cosmos/azure_data_cosmos/tests/framework/test_data.rs +++ b/sdk/cosmos/azure_data_cosmos/tests/framework/test_data.rs @@ -65,8 +65,9 @@ pub async fn create_container_with_items( let container_client = db.container_client("TestContainer").await?; for item in items { + let item_id = item.id.clone(); container_client - .create_item(item.partition_key.clone(), item, None) + .create_item(item.partition_key.clone(), &item_id, item, None) .await?; } diff --git a/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write.rs b/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write.rs index 86afec06471..ffb40180cb2 100644 --- a/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write.rs +++ b/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write.rs @@ -90,6 +90,7 @@ fn create_container_and_write_item<'a>( // This upsert operation triggers a routing decision log in the driver container_client .upsert_item( + "item1", "item1", &serde_json::json!({"id": "item1", "value": "test"}), None, diff --git a/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write_fault_injection.rs b/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write_fault_injection.rs index ad151e362c1..95de25d2c72 100644 --- a/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write_fault_injection.rs +++ b/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write_fault_injection.rs @@ -80,7 +80,9 @@ async fn verify_read_fails_with_injected_error( let pk = format!("Partition1-{}", unique_id); let item_id = format!("Item1-{}", unique_id); - container_client.create_item(&pk, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() @@ -231,7 +233,9 @@ pub async fn item_read_succeeds_when_fault_targets_create_item() -> Result<(), B let item_id = format!("Item1-{}", unique_id); // Create the item using the normal client (this should succeed) - container_client.create_item(&pk, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() @@ -314,7 +318,9 @@ pub async fn fault_injection_read_region_retry_503() -> Result<(), Box Result<() bool_value: true, }; let pk = format!("Partition-{}", unique_id); + let item_id = format!("Item-{}", unique_id); // Transport-generated 503 on a non-idempotent write (upsert) should NOT // be retried — the driver cannot know if the server processed the request. - let result = fault_container_client.upsert_item(&pk, &item, None).await; + let result = fault_container_client + .upsert_item(&pk, &item_id, &item, None) + .await; assert!( result.is_err(), @@ -476,7 +485,9 @@ pub async fn fault_injection_read_region_retry_404_1002() -> Result<(), Box Result<(), Box bool_value: true, }; let pk = format!("Partition-{}", unique_id); + let item_id = format!("Item-{}", unique_id); let response = fault_container_client - .create_item(&pk, &item, None) + .create_item(&pk, &item_id, &item, None) .await .expect("write should succeed via failover to satellite"); @@ -648,7 +660,9 @@ pub async fn fault_injection_read_connection_error_failover() -> Result<(), Box< let item_id = format!("Item-{}", unique_id); // Create item with the normal client - container_client.create_item(&pk, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() @@ -739,8 +753,11 @@ pub async fn fault_injection_write_response_timeout_does_not_retry() -> Result<( bool_value: true, }; let pk = format!("Partition-{}", unique_id); + let item_id = format!("Item-{}", unique_id); - let result = fault_container_client.create_item(&pk, &item, None).await; + let result = fault_container_client + .create_item(&pk, &item_id, &item, None) + .await; assert!( result.is_err(), @@ -807,7 +824,9 @@ pub async fn fault_injection_read_response_timeout_retries_to_satellite( let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() @@ -898,9 +917,10 @@ pub async fn fault_injection_connection_error_reverse_failover() -> Result<(), B bool_value: true, }; let pk = format!("Partition-{}", unique_id); + let item_id = format!("Item-{}", unique_id); let response = fault_container_client - .create_item(&pk, &item, None) + .create_item(&pk, &item_id, &item, None) .await .expect("write should succeed via reverse failover to hub"); @@ -971,7 +991,9 @@ pub async fn fault_injection_connection_error_local_retry_succeeds() -> Result<( let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() diff --git a/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write_retry_policies.rs b/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write_retry_policies.rs index 838147408c6..7cba26d28f1 100644 --- a/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write_retry_policies.rs +++ b/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write_retry_policies.rs @@ -93,7 +93,9 @@ pub async fn read_cross_region_retry_on_408() -> Result<(), Box> { let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() @@ -186,9 +188,12 @@ pub async fn write_no_cross_region_retry_on_408() -> Result<(), Box> bool_value: true, }; let pk = format!("Partition-{}", unique_id); + let item_id = format!("Item-{}", unique_id); // Write should fail with 408 — no cross-region retry for writes - let result = fault_container_client.create_item(&pk, &item, None).await; + let result = fault_container_client + .create_item(&pk, &item_id, &item, None) + .await; let err = result.expect_err("write should fail with 408 and not retry across regions"); assert_eq!( @@ -261,9 +266,12 @@ pub async fn upsert_no_cross_region_retry_on_408() -> Result<(), Box> bool_value: true, }; let pk = format!("Partition-{}", unique_id); + let item_id = format!("Item-{}", unique_id); // Upsert should fail with 408 — no cross-region retry for writes - let result = fault_container_client.upsert_item(&pk, &item, None).await; + let result = fault_container_client + .upsert_item(&pk, &item_id, &item, None) + .await; let err = result.expect_err("upsert should fail with 408 and not retry across regions"); assert_eq!( @@ -333,8 +341,11 @@ pub async fn query_cross_region_retry_on_408() -> Result<(), Box> { bool_value: true, }; let pk = format!("Partition-{}", unique_id); + let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() @@ -417,7 +428,9 @@ pub async fn read_cross_region_retry_on_500() -> Result<(), Box> { let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() @@ -507,7 +520,9 @@ pub async fn replace_no_cross_region_retry_on_408() -> Result<(), Box let item_id = format!("Item-{}", unique_id); // Create the item first via the non-fault client - container_client.create_item(&pk, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() @@ -599,7 +614,9 @@ pub async fn delete_no_cross_region_retry_on_408() -> Result<(), Box> let item_id = format!("Item-{}", unique_id); // Create the item first via the non-fault client - container_client.create_item(&pk, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() diff --git a/sdk/cosmos/azure_data_cosmos_benchmarks/src/lib.rs b/sdk/cosmos/azure_data_cosmos_benchmarks/src/lib.rs index 6487f1b5040..a52b728396b 100644 --- a/sdk/cosmos/azure_data_cosmos_benchmarks/src/lib.rs +++ b/sdk/cosmos/azure_data_cosmos_benchmarks/src/lib.rs @@ -303,14 +303,15 @@ pub async fn setup_live() -> (Arc, ItemReference) { // Create the benchmark item if it doesn't already exist. let item_body = format!(r#"{{"id": "{}", "pk": "{}"}}"#, item_id, pk_value); + let item_ref = ItemReference::from_name( + &container_ref, + PartitionKey::from(pk_value.clone()), + item_id.clone(), + ); ignore_conflict( driver .execute_operation( - CosmosOperation::create_item( - container_ref.clone(), - PartitionKey::from(pk_value.clone()), - ) - .with_body(item_body.into_bytes()), + CosmosOperation::create_item(item_ref).with_body(item_body.into_bytes()), OperationOptions::default(), ) .await, diff --git a/sdk/cosmos/azure_data_cosmos_driver/src/driver/pipeline/operation_pipeline.rs b/sdk/cosmos/azure_data_cosmos_driver/src/driver/pipeline/operation_pipeline.rs index d4cd938e5c0..e179abd85ed 100644 --- a/sdk/cosmos/azure_data_cosmos_driver/src/driver/pipeline/operation_pipeline.rs +++ b/sdk/cosmos/azure_data_cosmos_driver/src/driver/pipeline/operation_pipeline.rs @@ -444,9 +444,7 @@ fn build_transport_request( custom_headers: Option<&std::collections::HashMap>, ctx: &TransportRequestContext<'_>, ) -> azure_core::Result { - let resource_ref = operation.resource_reference(); - // Compute both paths in a single pass with a single allocation. - let paths = resource_ref.compute_paths(); + let paths = operation.compute_resource_paths(); let url = { let mut base = ctx.routing.selected_url.clone(); let request_path = paths.request_path(); @@ -1249,8 +1247,9 @@ mod tests { #[test] fn build_transport_request_sets_is_upsert_header() { - let operation = CosmosOperation::upsert_item(test_container(), PartitionKey::from("pk1")) - .with_body(b"{}".to_vec()); + let container = test_container(); + let item = ItemReference::from_name(&container, PartitionKey::from("pk1"), "doc1"); + let operation = CosmosOperation::upsert_item(item).with_body(b"{}".to_vec()); let routing = test_routing(); let activity_id = ActivityId::from_string("default-activity".to_string()); @@ -1270,12 +1269,20 @@ mod tests { .get_optional_str(&HeaderName::from_static("x-ms-documentdb-is-upsert")) .expect("is-upsert header should be set"); assert_eq!(is_upsert, "true"); + + // Upsert targets the collection feed URL, not the individual document. + assert_eq!( + request.url.path(), + "/dbs/testdb/colls/testcontainer/docs", + "upsert should POST to the collection feed, not /docs/doc1" + ); } #[test] fn build_transport_request_omits_is_upsert_header_for_create() { - let operation = CosmosOperation::create_item(test_container(), PartitionKey::from("pk1")) - .with_body(b"{}".to_vec()); + let container = test_container(); + let item = ItemReference::from_name(&container, PartitionKey::from("pk1"), "doc1"); + let operation = CosmosOperation::create_item(item).with_body(b"{}".to_vec()); let routing = test_routing(); let activity_id = ActivityId::from_string("default-activity".to_string()); @@ -1297,6 +1304,13 @@ mod tests { .is_none(), "is-upsert header should not be set for create" ); + + // Create targets the collection feed URL, not the individual document. + assert_eq!( + request.url.path(), + "/dbs/testdb/colls/testcontainer/docs", + "create should POST to the collection feed, not /docs/doc1" + ); } #[test] diff --git a/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_operation.rs b/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_operation.rs index a474d2061d1..0df26573147 100644 --- a/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_operation.rs +++ b/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_operation.rs @@ -86,6 +86,24 @@ impl CosmosOperation { &self.resource_reference } + /// Computes the request path and signing link for this operation. + /// + /// Create and Upsert document operations use feed-style paths (targeting + /// the collection URL) even though they carry an item id, because the + /// Cosmos DB REST API POSTs these to the collection feed. All other + /// operations use the standard resource paths. + pub(crate) fn compute_resource_paths(&self) -> crate::models::ResourcePaths { + if matches!( + self.operation_type, + OperationType::Create | OperationType::Upsert + ) && self.resource_type == ResourceType::Document + { + self.resource_reference.compute_feed_paths() + } else { + self.resource_reference.compute_paths() + } + } + /// Returns the container for this operation, if applicable. /// /// Returns `None` for account-level and database-level operations. @@ -376,9 +394,10 @@ impl CosmosOperation { // ===== Data Plane Factory Methods ===== - /// Creates an item (document) in a container. + /// Creates a new item (document) in a container. /// - /// The `container` and `partition_key` identify where to create the document. + /// The `ItemReference` contains the container, partition key, and item identifier, + /// providing all the information needed for the operation. /// Use `with_body()` to provide the document JSON. /// /// # Example @@ -386,7 +405,7 @@ impl CosmosOperation { /// ```no_run /// use azure_data_cosmos_driver::driver::CosmosDriverRuntime; /// use azure_data_cosmos_driver::models::{ - /// AccountReference, CosmosOperation, ContainerReference, PartitionKey, + /// AccountReference, CosmosOperation, ItemReference, PartitionKey, /// }; /// use azure_data_cosmos_driver::options::OperationOptions; /// use url::Url; @@ -400,10 +419,10 @@ impl CosmosOperation { /// let driver = runtime.get_or_create_driver(account, None).await?; /// let container = driver.resolve_container("my-database", "my-container").await?; /// - /// let pk = PartitionKey::from("pk-value"); + /// let item = ItemReference::from_name(&container, PartitionKey::from("pk-value"), "doc1"); /// let result = driver /// .execute_operation( - /// CosmosOperation::create_item(container, pk) + /// CosmosOperation::create_item(item) /// .with_body(br#"{"id": "doc1", "pk": "pk-value", "data": "hello"}"#.to_vec()), /// OperationOptions::default(), /// ) @@ -411,11 +430,9 @@ impl CosmosOperation { /// # Ok(()) /// # } /// ``` - pub fn create_item(container: ContainerReference, partition_key: PartitionKey) -> Self { - let resource_ref: CosmosResourceReference = CosmosResourceReference::from(container) - .with_resource_type(ResourceType::Document) - .into_feed_reference(); - Self::new(OperationType::Create, resource_ref).with_partition_key(partition_key) + pub fn create_item(item: ItemReference) -> Self { + let partition_key = item.partition_key().clone(); + Self::new(OperationType::Create, item).with_partition_key(partition_key) } /// Reads an item (document) from a container. @@ -466,16 +483,13 @@ impl CosmosOperation { /// Upserts (creates or replaces) an item (document) in a container. /// - /// The Cosmos DB REST API treats upsert as a `POST` to the collection feed - /// (same as create), so this takes a `ContainerReference` and `PartitionKey` - /// rather than an `ItemReference`. + /// The `ItemReference` contains the container, partition key, and item identifier, + /// providing all the information needed for the operation. /// Use `with_body()` to provide the document JSON. /// If an item with the same ID exists, it will be replaced; otherwise, a new item is created. - pub fn upsert_item(container: ContainerReference, partition_key: PartitionKey) -> Self { - let resource_ref: CosmosResourceReference = CosmosResourceReference::from(container) - .with_resource_type(ResourceType::Document) - .into_feed_reference(); - Self::new(OperationType::Upsert, resource_ref).with_partition_key(partition_key) + pub fn upsert_item(item: ItemReference) -> Self { + let partition_key = item.partition_key().clone(); + Self::new(OperationType::Upsert, item).with_partition_key(partition_key) } /// Replaces an existing item (document) in a container. diff --git a/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_resource_reference.rs b/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_resource_reference.rs index 8f909d89a01..cb6fc035ac2 100644 --- a/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_resource_reference.rs +++ b/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_resource_reference.rs @@ -35,8 +35,8 @@ pub(crate) struct ResourcePaths { buf: String, /// Byte index in `buf` where the signing link ends (exclusive). /// - /// For non-feed: `buf.len()` → signing link = `buf[1..]` - /// For feed: `parent.len()` → signing link = `buf[1..signing_end]` + /// For non-feed: `buf.len()` → signing link = `buf[1..]` + /// For feed: `parent.len()` → signing link = `buf[1..signing_end]` /// Always `>= 1` when `buf` is non-empty (skips the leading `/`). signing_end: usize, /// Signing link override for offer resources. @@ -158,6 +158,28 @@ impl CosmosResourceReference { self } + /// Computes paths treating this reference as a feed operation. + /// + /// Used by Create and Upsert which carry an [`ItemReference`] (with an + /// item id) but still POST to the parent (collection) URL and sign + /// against the parent resource. + pub(crate) fn compute_feed_paths(&self) -> ResourcePaths { + // Temporarily treat the reference as a feed for path computation. + let parent = self.parent_link_cow(); + let segment = self.resource_type.path_segment(); + let buf = if parent.is_empty() { + format!("/{}", segment) + } else { + format!("{}/{}", parent, segment) + }; + let signing_end = if parent.is_empty() { 1 } else { parent.len() }; + ResourcePaths { + buf, + signing_end, + signing_override: None, + } + } + /// Returns the resource link used for authorization signing. /// /// For feed operations this is the **parent** resource's link (because @@ -762,6 +784,35 @@ mod tests { assert_compute_paths_consistent(&r); } + #[test] + fn compute_feed_paths_item_reference() { + // An ItemReference carries the document id, but compute_feed_paths + // must produce the same feed-style paths as compute_paths on a + // feed reference (without the item id in the URL). + let item = ItemReference::from_name(&test_container(), PartitionKey::from("pk1"), "doc1"); + let r: CosmosResourceReference = item.into(); + + let feed_paths = r.compute_feed_paths(); + assert_eq!( + feed_paths.request_path(), + "/dbs/testdb/colls/testcontainer/docs", + "request path should target the collection feed, not the individual document" + ); + assert_eq!( + feed_paths.signing_link(), + "dbs/testdb/colls/testcontainer", + "signing link should be the parent container path" + ); + + // Verify consistency with compute_paths on an equivalent feed reference. + let feed_ref = CosmosResourceReference::from(test_container()) + .with_resource_type(ResourceType::Document) + .into_feed_reference(); + let expected = feed_ref.compute_paths(); + assert_eq!(feed_paths.request_path(), expected.request_path()); + assert_eq!(feed_paths.signing_link(), expected.signing_link()); + } + #[test] fn compute_paths_offer_uses_signing_override() { let account = test_account(); diff --git a/sdk/cosmos/azure_data_cosmos_driver/tests/emulator_tests/driver_fault_injection.rs b/sdk/cosmos/azure_data_cosmos_driver/tests/emulator_tests/driver_fault_injection.rs index de3810bc294..439377273ab 100644 --- a/sdk/cosmos/azure_data_cosmos_driver/tests/emulator_tests/driver_fault_injection.rs +++ b/sdk/cosmos/azure_data_cosmos_driver/tests/emulator_tests/driver_fault_injection.rs @@ -42,7 +42,7 @@ pub async fn fault_injection_probability_zero_never_fails() -> Result<(), Box Result<(), .await?; let item_json = br#"{"id": "item1", "pk": "pk1", "value": "test"}"#; - context.create_item(&container, "pk1", item_json).await?; + context + .create_item(&container, "item1", "pk1", item_json) + .await?; // With probability 1.0, the read should fail let read_result = context.read_item(&container, "item1", "pk1").await; @@ -163,7 +165,7 @@ pub async fn fault_injection_operation_type_filter() -> Result<(), Box Result<(), Box< .await?; let item_json = br#"{"id": "item1", "pk": "pk1", "value": "test"}"#; - context.create_item(&container, "pk1", item_json).await?; + context.create_item(&container, "item1", "pk1", item_json).await?; // Execute reads to consume the hit limit. // Due to internal retries, the limit may be exhausted within fewer @@ -305,7 +307,9 @@ pub async fn fault_injection_connection_error() -> Result<(), Box> { .await?; let item_json = br#"{"id": "item1", "pk": "pk1", "value": "test"}"#; - context.create_item(&container, "pk1", item_json).await?; + context + .create_item(&container, "item1", "pk1", item_json) + .await?; // With a connection error injected, the read should fail let read_result = context.read_item(&container, "item1", "pk1").await; diff --git a/sdk/cosmos/azure_data_cosmos_driver/tests/emulator_tests/driver_item_operations.rs b/sdk/cosmos/azure_data_cosmos_driver/tests/emulator_tests/driver_item_operations.rs index dca95e9ab52..36e83165271 100644 --- a/sdk/cosmos/azure_data_cosmos_driver/tests/emulator_tests/driver_item_operations.rs +++ b/sdk/cosmos/azure_data_cosmos_driver/tests/emulator_tests/driver_item_operations.rs @@ -54,7 +54,7 @@ pub async fn create_and_read_item() -> Result<(), Box> { // Create the item let create_result = context - .create_item(&container, item.pk.clone(), &item_json) + .create_item(&container, &item.id, item.pk.clone(), &item_json) .await?; // Validate create diagnostics @@ -128,7 +128,7 @@ pub async fn control_plane_uses_metadata_pipeline() -> Result<(), Box let item_json = serde_json::to_vec(&test_item)?; let result = context - .create_item(&container, test_item.pk.clone(), &item_json) + .create_item(&container, &test_item.id, test_item.pk.clone(), &item_json) .await?; // Verify item creation succeeded @@ -164,7 +164,7 @@ pub async fn diagnostics_contain_expected_fields() -> Result<(), Box> let item_json = serde_json::to_vec(&item)?; let result = context - .create_item(&container, item.pk.clone(), &item_json) + .create_item(&container, &item.id, item.pk.clone(), &item_json) .await?; let diagnostics = result.diagnostics(); diff --git a/sdk/cosmos/azure_data_cosmos_driver/tests/framework/test_client.rs b/sdk/cosmos/azure_data_cosmos_driver/tests/framework/test_client.rs index db1f5bb65fa..6657157d9e2 100644 --- a/sdk/cosmos/azure_data_cosmos_driver/tests/framework/test_client.rs +++ b/sdk/cosmos/azure_data_cosmos_driver/tests/framework/test_client.rs @@ -338,6 +338,7 @@ impl DriverTestRunContext { pub async fn create_item( &self, container: &ContainerReference, + item_id: &str, partition_key: impl Into, body: &[u8], ) -> Result> { @@ -348,8 +349,8 @@ impl DriverTestRunContext { .await?; let pk = partition_key.into(); - let operation = - CosmosOperation::create_item(container.clone(), pk).with_body(body.to_vec()); + let item_ref = ItemReference::from_name(container, pk, item_id.to_owned()); + let operation = CosmosOperation::create_item(item_ref).with_body(body.to_vec()); let result = driver .execute_operation(operation, OperationOptions::default()) diff --git a/sdk/cosmos/azure_data_cosmos_perf/src/operations/create_item.rs b/sdk/cosmos/azure_data_cosmos_perf/src/operations/create_item.rs index 54602c223fa..7af0a13ee8b 100644 --- a/sdk/cosmos/azure_data_cosmos_perf/src/operations/create_item.rs +++ b/sdk/cosmos/azure_data_cosmos_perf/src/operations/create_item.rs @@ -49,7 +49,7 @@ impl Operation for CreateItemOperation { }; container - .create_item(&item.partition_key, &item, self.options.clone()) + .create_item(&item.partition_key, &id, &item, self.options.clone()) .await?; self.items.push(SeededItem { id, partition_key }); diff --git a/sdk/cosmos/azure_data_cosmos_perf/src/operations/upsert_item.rs b/sdk/cosmos/azure_data_cosmos_perf/src/operations/upsert_item.rs index 6338a08c9bb..0b804483d3f 100644 --- a/sdk/cosmos/azure_data_cosmos_perf/src/operations/upsert_item.rs +++ b/sdk/cosmos/azure_data_cosmos_perf/src/operations/upsert_item.rs @@ -44,7 +44,7 @@ impl Operation for UpsertItemOperation { }; container - .upsert_item(&item.partition_key, &item, self.options.clone()) + .upsert_item(&item.partition_key, &seeded.id, &item, self.options.clone()) .await?; Ok(()) } diff --git a/sdk/cosmos/azure_data_cosmos_perf/src/runner.rs b/sdk/cosmos/azure_data_cosmos_perf/src/runner.rs index fdda04c78f8..1d24329782e 100644 --- a/sdk/cosmos/azure_data_cosmos_perf/src/runner.rs +++ b/sdk/cosmos/azure_data_cosmos_perf/src/runner.rs @@ -404,7 +404,7 @@ async fn upsert_results( }; if let Err(e) = container - .upsert_item(&result.partition_key, &result, None) + .upsert_item(&result.partition_key, &result.id, &result, None) .await { eprintln!("Warning: failed to upsert perf result: {e}"); @@ -439,7 +439,10 @@ async fn upsert_error( error_message: format!("{error}"), source_message: error_source_chain(error), }; - if let Err(e) = container.upsert_item(&doc.partition_key, &doc, None).await { + if let Err(e) = container + .upsert_item(&doc.partition_key, &id, &doc, None) + .await + { eprintln!("Warning: failed to upsert error result: {e}"); } } diff --git a/sdk/cosmos/azure_data_cosmos_perf/src/seed.rs b/sdk/cosmos/azure_data_cosmos_perf/src/seed.rs index 917e8b22af4..f3ab9844836 100644 --- a/sdk/cosmos/azure_data_cosmos_perf/src/seed.rs +++ b/sdk/cosmos/azure_data_cosmos_perf/src/seed.rs @@ -103,7 +103,7 @@ pub async fn seed_container( }; let result = container - .upsert_item(&item.partition_key, &item, None) + .upsert_item(&item.partition_key, &item.id, &item, None) .await; if result.is_ok() {