Fix upsert create item regression (#4280)

## What

This PR makes `create_item` and `upsert_item` require an explicit item
id, matching the pattern already used by `read_item`, `replace_item`,
and `delete_item`.

### Breaking Changes

**`azure_data_cosmos` (SDK):**
- `ContainerClient::create_item()` signature changed from
`(partition_key, item, options)` to `(partition_key, item_id, item,
options)`
- `ContainerClient::upsert_item()` signature changed from
`(partition_key, item, options)` to `(partition_key, item_id, item,
options)`

**`azure_data_cosmos_driver` (internal):**
- `CosmosOperation::create_item()` now takes `ItemReference` instead of
`(ContainerReference, PartitionKey)`
- `CosmosOperation::upsert_item()` now takes `ItemReference` instead of
`(ContainerReference, PartitionKey)`

All five point operations (`create_item`, `upsert_item`, `read_item`,
`replace_item`, `delete_item`) now follow the same pattern in both the
SDK and the driver.

## Why

PR #4128 cut over more point operations to the driver but inadvertently
removed the `ItemReference`-based overloads for `create_item` and
`upsert_item`. This was a regression — the original design requires the
caller to provide the item id explicitly so the driver never needs to
parse the request body to extract the document id. Parsing the body in
the driver is both unnecessary overhead and a layering violation (the
SDK should own serialization concerns).

## How

- **Driver (`cosmos_operation.rs`):** `create_item` and `upsert_item`
factory methods now take `ItemReference` and call
`Self::new(OperationType::Create/Upsert, item)`, identical to
`read_item`/`delete_item`/`replace_item`.
- **Driver (`operation_pipeline.rs`):** `build_transport_request`
detects Create/Upsert Document operations and uses
`compute_feed_paths()` instead of `compute_paths()`, since these
operations POST to the collection feed URL even though the operation
carries the item id.
- **Driver (`cosmos_resource_reference.rs`):** Added
`compute_feed_paths()` to compute feed-style paths (parent URL + signing
link) for references that carry a leaf id.
- **SDK (`container_client.rs`):** `create_item` and `upsert_item` now
accept `item_id: &str` and construct an `ItemReference` before
delegating to the driver.
- All tests, examples, benchmarks, and perf tooling updated.

## Testing

Existing test coverage is sufficient

Author: FabianMeiswinkel

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 @@
       ]
     }
   ]
-}
+}

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<ThroughputProperties>` 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))
 

sdk/cosmos/azure_data_cosmos/README.md

@@ -108,7 +108,7 @@ async fn example(cosmos_client: CosmosClient) -> Result<(), Box<dyn std::error::
     let container = cosmos_client.database_client("myDatabase").container_client("myContainer").await?;
 
     // Create an item
-    container.create_item("partition1", item, None).await?;
+    container.create_item("partition1", "1", item, None).await?;
 
     // Read an item
     let item_response = container.read_item("partition1", "1", None).await?;

sdk/cosmos/azure_data_cosmos/examples/cosmos/create.rs

@@ -33,6 +33,10 @@ pub enum Subcommands {
         #[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,
@@ -77,6 +81,7 @@ impl CreateCommand {
                 database,
                 container,
                 partition_key,
+                item_id,
                 json,
                 show_updated,
             } => {
@@ -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");
 

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 {

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::<Product>();
     /// # Ok(())
@@ -317,16 +318,22 @@ impl ContainerClient {
     pub async fn create_item<T: Serialize>(
         &self,
         partition_key: impl Into<PartitionKey>,
+        item_id: &str,
         item: T,
         options: Option<ItemWriteOptions>,
     ) -> azure_core::Result<ItemResponse<()>> {
         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,24 +510,30 @@ 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::<Product>()?;
     /// Ok(())
     /// # }
     pub async fn upsert_item<T: Serialize>(
         &self,
         partition_key: impl Into<PartitionKey>,
+        item_id: &str,
         item: T,
         options: Option<ItemWriteOptions>,
     ) -> azure_core::Result<ItemResponse<()>> {
         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::<serde_json::Value>("", todo!(), todo!()));
+        assert_send(client.create_item::<serde_json::Value>("", todo!(), todo!(), todo!()));
         assert_send(client.replace_item::<serde_json::Value>("", todo!(), todo!(), todo!()));
-        assert_send(client.upsert_item::<serde_json::Value>("", todo!(), todo!()));
+        assert_send(client.upsert_item::<serde_json::Value>("", todo!(), todo!(), todo!()));
         assert_send(client.read_item::<serde_json::Value>("", todo!(), todo!()));
         assert_send(client.delete_item("", todo!(), todo!()));
         assert_send(client.execute_transactional_batch(todo!(), todo!()));

sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_batch.rs

@@ -132,10 +132,10 @@ pub async fn batch_mixed_operations() -> Result<(), Box<dyn Error>> {
             };
 
             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<dyn Error>> {
             };
 
             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)