[Internal] DTS: Adds distributed read transaction API to CosmosClient (#5795)

# Pull Request Template

## Description

This pull request introduces support for distributed read transactions
in the Cosmos DB SDK, allowing users to perform multiple read operations
across partitions and containers atomically. The main changes include
the addition of new classes and methods to create and manage distributed
read transactions, comprehensive tests, and supporting utilities for
test scenarios.

**Distributed Read Transaction Feature:**

* Added a new abstract class `DistributedReadTransaction` that allows
chaining multiple read operations and committing them atomically. This
class is obtained via the new
`CosmosClient.CreateDistributedReadTransaction()` method.
* Implemented the concrete class `DistributedReadTransactionCore`, which
manages the list of read operations, validates inputs, and delegates
transaction execution to the committer.

**Public API and Entry Point:**

* Added the `CreateDistributedReadTransaction()` method to
`CosmosClient`, providing a public (or internal, depending on build)
entry point for distributed read transactions.

## Type of change

Please delete options that are not relevant.

- [] New feature (non-breaking change which adds functionality)

## Closing issues

To automatically close an issue: closes #IssueNumber

---------

Co-authored-by: Kiran Kumar Kolli <kirankk@microsoft.com>

Author: Meghana-Palaparthi

Microsoft.Azure.Cosmos/src/CosmosClient.cs

@@ -1177,6 +1177,20 @@ virtual DistributedWriteTransaction CreateDistributedWriteTransaction()
             return new DistributedWriteTransactionCore(this.ClientContext);
         }
 
+        /// <summary>
+        /// Creates a new instance of a distributed read transaction.
+        /// </summary>
+        /// <returns>An instance of <see cref="DistributedReadTransaction"/>.</returns>
+#if INTERNAL
+        public
+#else
+        internal
+#endif
+        virtual DistributedReadTransaction CreateDistributedReadTransaction()
+        {
+            return new DistributedReadTransactionCore(this.ClientContext);
+        }
+
         /// <summary>
         /// Send a request for creating a database.
         ///

Microsoft.Azure.Cosmos/src/DistributedTransaction/DistributedReadTransaction.cs

@@ -0,0 +1,40 @@
+// ------------------------------------------------------------
+// Copyright (c) Microsoft Corporation.  All rights reserved.
+// ------------------------------------------------------------
+
+namespace Microsoft.Azure.Cosmos
+{
+    using System.IO;
+
+    /// <summary>
+    /// Represents a distributed transaction that supports read operations across multiple partitions and containers.
+    /// </summary>
+    /// <remarks>
+    /// Use <see cref="CosmosClient.CreateDistributedReadTransaction"/> to obtain an instance.
+    /// Add read operations using <see cref="ReadItem"/> then call
+    /// <see cref="DistributedTransaction.CommitTransactionAsync"/> to execute all reads atomically.
+    /// </remarks>
+#if INTERNAL
+    public
+#else
+    internal
+#endif
+    abstract class DistributedReadTransaction : DistributedTransaction
+    {
+        /// <summary>
+        /// Adds a read operation to the distributed transaction.
+        /// </summary>
+        /// <param name="database">The name of the database containing the container.</param>
+        /// <param name="collection">The name of the container where the item exists.</param>
+        /// <param name="partitionKey">The partition key for the item.</param>
+        /// <param name="id">The unique identifier of the item to read.</param>
+        /// <param name="requestOptions">Options for the read operation.</param>
+        /// <returns>The current <see cref="DistributedReadTransaction"/> instance for method chaining.</returns>
+        public abstract DistributedReadTransaction ReadItem(
+            string database,
+            string collection,
+            PartitionKey partitionKey,
+            string id,
+            DistributedTransactionRequestOptions requestOptions = null);
+    }
+}

Microsoft.Azure.Cosmos/src/DistributedTransaction/DistributedReadTransactionCore.cs

@@ -0,0 +1,78 @@
+// ------------------------------------------------------------
+// Copyright (c) Microsoft Corporation.  All rights reserved.
+// ------------------------------------------------------------
+
+namespace Microsoft.Azure.Cosmos
+{
+    using System;
+    using System.Collections.Generic;
+    using System.Threading;
+    using System.Threading.Tasks;
+    using Microsoft.Azure.Documents;
+
+    internal class DistributedReadTransactionCore : DistributedReadTransaction
+    {
+        private readonly CosmosClientContext clientContext;
+        private readonly List<DistributedTransactionOperation> operations;
+
+        internal DistributedReadTransactionCore(CosmosClientContext clientContext)
+        {
+            this.clientContext = clientContext ?? throw new ArgumentNullException(nameof(clientContext));
+            this.operations = new List<DistributedTransactionOperation>();
+        }
+
+        public override DistributedReadTransaction ReadItem(
+            string database,
+            string collection,
+            PartitionKey partitionKey,
+            string id,
+            DistributedTransactionRequestOptions requestOptions = null)
+        {
+            DistributedReadTransactionCore.ValidateContainerReference(database, collection);
+            DistributedReadTransactionCore.ValidateItemId(id);
+
+            this.operations.Add(
+                new DistributedTransactionOperation(
+                    operationType: OperationType.Read,
+                    operationIndex: this.operations.Count,
+                    database: database,
+                    container: collection,
+                    partitionKey: partitionKey,
+                    id: id,
+                    requestOptions: requestOptions));
+
+            return this;
+        }
+
+        public override async Task<DistributedTransactionResponse> CommitTransactionAsync(
+            CancellationToken cancellationToken = default)
+        {
+            DistributedTransactionCommitter committer = new DistributedTransactionCommitter(
+                operations: this.operations,
+                clientContext: this.clientContext);
+
+            return await committer.CommitTransactionAsync(cancellationToken);
+        }
+
+        private static void ValidateContainerReference(string database, string collection)
+        {
+            if (string.IsNullOrWhiteSpace(database))
+            {
+                throw new ArgumentNullException(nameof(database));
+            }
+
+            if (string.IsNullOrWhiteSpace(collection))
+            {
+                throw new ArgumentNullException(nameof(collection));
+            }
+        }
+
+        private static void ValidateItemId(string id)
+        {
+            if (string.IsNullOrWhiteSpace(id))
+            {
+                throw new ArgumentNullException(nameof(id));
+            }
+        }
+    }
+}

Microsoft.Azure.Cosmos/src/DistributedTransaction/DistributedTransactionOperation.cs

@@ -57,7 +57,9 @@ public DistributedTransactionOperation(
 
         internal string SessionToken { get; set; }
 
-        internal string ETag => this.RequestOptions?.IfMatchEtag;
+        internal string ETag => this.OperationType == OperationType.Read
+            ? this.RequestOptions?.IfNoneMatchEtag
+            : this.RequestOptions?.IfMatchEtag;
 
         internal Stream ResourceStream { get; set; }
 

Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.EmulatorTests/DistributedTransaction/DistributedTransactionTests.cs

@@ -784,6 +784,153 @@ public async Task ValidateSessionTokenMergedIntoDtcClient()
             }
         }
 
+        // Read Transaction Tests
+
+        [TestMethod]
+        public async Task ValidateReadTransactionHappyPath()
+        {
+            // Arrange
+            ToDoActivity doc1 = ToDoActivity.CreateRandomToDoActivity();
+            ToDoActivity doc2 = ToDoActivity.CreateRandomToDoActivity();
+
+            DistributedTransactionMockHandler handler = new DistributedTransactionMockHandler(
+                request => Task.FromResult(this.BuildMockResponse(
+                    HttpStatusCode.OK,
+                    BuildReadSuccessResponseJson(2, JsonSerializer.Serialize(doc1), JsonSerializer.Serialize(doc2)))));
+
+            using CosmosClient client = this.CreateMockClient(handler);
+
+            // Act
+            DistributedTransactionResponse response = await client
+                .CreateDistributedReadTransaction()
+                .ReadItem(this.database.Id, this.container.Id, new PartitionKey(doc1.pk), doc1.id)
+                .ReadItem(this.database.Id, this.container.Id, new PartitionKey(doc2.pk), doc2.id)
+                .CommitTransactionAsync(CancellationToken.None);
+
+            // Assert
+            Assert.IsNotNull(handler.CapturedRequestBody);
+            Assert.AreEqual(HttpStatusCode.OK, response.StatusCode);
+            Assert.IsTrue(response.IsSuccessStatusCode);
+            Assert.AreEqual(2, response.Count);
+
+            response.Dispose();
+        }
+
+        [TestMethod]
+        public async Task ValidateReadTransactionRequestStructure()
+        {
+            // Arrange
+            ToDoActivity doc = ToDoActivity.CreateRandomToDoActivity();
+
+            DistributedTransactionMockHandler handler = new DistributedTransactionMockHandler(
+                request => Task.FromResult(this.BuildMockResponse(
+                    HttpStatusCode.OK,
+                    BuildReadSuccessResponseJson(1, JsonSerializer.Serialize(doc)))));
+
+            using CosmosClient client = this.CreateMockClient(handler);
+
+            // Act
+            DistributedTransactionResponse response = await client
+                .CreateDistributedReadTransaction()
+                .ReadItem(this.database.Id, this.container.Id, new PartitionKey(doc.pk), doc.id)
+                .CommitTransactionAsync(CancellationToken.None);
+
+            // Assert – request structure
+            Assert.IsNotNull(handler.CapturedRequestBody);
+            using JsonDocument requestJson = JsonDocument.Parse(handler.CapturedRequestBody);
+            JsonElement operation = requestJson.RootElement.GetProperty("operations")[0];
+
+            Assert.AreEqual(OperationType.Read.ToString(), operation.GetProperty("operationType").GetString());
+            Assert.AreEqual(doc.id, operation.GetProperty("id").GetString());
+            Assert.IsTrue(operation.TryGetProperty("databaseName", out _), "databaseName should be present");
+            Assert.IsTrue(operation.TryGetProperty("collectionName", out _), "collectionName should be present");
+            Assert.IsTrue(operation.TryGetProperty("partitionKey", out _), "partitionKey should be present");
+            Assert.IsFalse(operation.TryGetProperty("resourceBody", out _), "resourceBody must NOT be present for read operations");
+
+            response.Dispose();
+        }
+
+        [TestMethod]
+        public async Task ValidateReadTransactionResponseDeserialization()
+        {
+            // Arrange
+            ToDoActivity expectedDoc = ToDoActivity.CreateRandomToDoActivity();
+
+            DistributedTransactionMockHandler handler = new DistributedTransactionMockHandler(
+                request => Task.FromResult(this.BuildMockResponse(
+                    HttpStatusCode.OK,
+                    BuildReadSuccessResponseJson(1, JsonSerializer.Serialize(expectedDoc)))));
+
+            using CosmosClient client = this.CreateMockClient(handler);
+
+            // Act
+            DistributedTransactionResponse response = await client
+                .CreateDistributedReadTransaction()
+                .ReadItem(this.database.Id, this.container.Id, new PartitionKey(expectedDoc.pk), expectedDoc.id)
+                .CommitTransactionAsync(CancellationToken.None);
+
+            // Assert
+            Assert.IsTrue(response.IsSuccessStatusCode);
+            ToDoActivity actualDoc = JsonSerializer.Deserialize<ToDoActivity>(response[0].ResourceStream);
+            Assert.IsNotNull(actualDoc);
+            Assert.AreEqual(expectedDoc.id, actualDoc.id);
+            Assert.AreEqual(expectedDoc.pk, actualDoc.pk);
+            Assert.AreEqual(expectedDoc.taskNum, actualDoc.taskNum);
+
+            response.Dispose();
+        }
+
+        [TestMethod]
+        public async Task ValidateReadTransactionResourceStream()
+        {
+            // Arrange
+            ToDoActivity expectedDoc = ToDoActivity.CreateRandomToDoActivity();
+
+            DistributedTransactionMockHandler handler = new DistributedTransactionMockHandler(
+                request => Task.FromResult(this.BuildMockResponse(
+                    HttpStatusCode.OK,
+                    BuildReadSuccessResponseJson(1, JsonSerializer.Serialize(expectedDoc)))));
+
+            using CosmosClient client = this.CreateMockClient(handler);
+
+            // Act
+            DistributedTransactionResponse response = await client
+                .CreateDistributedReadTransaction()
+                .ReadItem(this.database.Id, this.container.Id, new PartitionKey(expectedDoc.pk), expectedDoc.id)
+                .CommitTransactionAsync(CancellationToken.None);
+
+            // Assert – raw stream access
+            Stream stream = response[0].ResourceStream;
+            Assert.IsNotNull(stream);
+            ToDoActivity actualDoc = JsonSerializer.Deserialize<ToDoActivity>(stream);
+            Assert.AreEqual(expectedDoc.id, actualDoc.id);
+            Assert.AreEqual(expectedDoc.pk, actualDoc.pk);
+
+            response.Dispose();
+        }
+
+        [TestMethod]
+        public void ValidateReadTransactionMissingIdThrows()
+        {
+            using CosmosClient client = this.CreateMockClient(new DistributedTransactionMockHandler(
+                request => Task.FromResult(this.BuildMockResponse(HttpStatusCode.OK, BuildSuccessResponseJson(1)))));
+
+            Assert.ThrowsException<ArgumentNullException>(() =>
+                client.CreateDistributedReadTransaction()
+                    .ReadItem(this.database.Id, this.container.Id, new PartitionKey("pk"), id: null));
+        }
+
+        [TestMethod]
+        public void ValidateReadTransactionMissingDatabaseThrows()
+        {
+            using CosmosClient client = this.CreateMockClient(new DistributedTransactionMockHandler(
+                request => Task.FromResult(this.BuildMockResponse(HttpStatusCode.OK, BuildSuccessResponseJson(1)))));
+
+            Assert.ThrowsException<ArgumentNullException>(() =>
+                client.CreateDistributedReadTransaction()
+                    .ReadItem(null, this.container.Id, new PartitionKey("pk"), "item-id"));
+        }
+
         // Helpers
 
         private void ValidateValueKind(JsonElement operation, string property, JsonValueKind expectedValueKind, int operationIndex, bool isRequired)
@@ -827,6 +974,18 @@ private static string BuildSuccessResponseJson(int operationCount)
             return $@"{{""operationResponses"":[{string.Join(",", results)}]}}";
         }
 
+        private static string BuildReadSuccessResponseJson(int operationCount, params string[] itemJsonBodies)
+        {
+            List<string> results = new List<string>();
+            for (int i = 0; i < operationCount; i++)
+            {
+                string body = i < itemJsonBodies.Length ? itemJsonBodies[i] : "{}";
+                results.Add($@"{{""index"":{i},""statusCode"":200,""etag"":""\""etag-{i}\"""",""resourceBody"":{body}}}");
+            }
+
+            return $@"{{""operationResponses"":[{string.Join(",", results)}]}}";
+        }
+
         // Mock handler
 
         /// <summary>