Change feed: Adds id and pk to ChangeFeedMetadata for delete operations

[yash2710]

Description

Adds id and partition key to ChangeFeedMetadata. This is populated for delete operations using all versions all deletes change feed mode so that when customers get the delete, they know what was deleted. Note it doesn't have the value of the document that was deleted only the id and partition key through which a document can be uniquely identified.
Please note that user provided serializer is not being used.

This PR includes commits from #5191 and additional simplifications and unit test changes

Type of change

New feature (non-breaking change which adds functionality)

Basic Usage

Single Partition Key Example

using Microsoft.Azure.Cosmos;

public class Product
{
    public string id { get; set; }
    public string categoryId { get; set; }
    public string name { get; set; }
    public decimal price { get; set; }
}

// Create Change Feed Processor with All Versions and Deletes mode
ChangeFeedProcessor processor = container
    .GetChangeFeedProcessorBuilderWithAllVersionsAndDeletes<Product>(
        processorName: "productProcessor",
        onChangesDelegate: async (context, changes, cancellationToken) =>
        {
            foreach (ChangeFeedItem<Product> change in changes)
            {
                switch (change.Metadata.OperationType)
                {
                    case ChangeFeedOperationType.Create:
                    case ChangeFeedOperationType.Replace:
                        Console.WriteLine($"Document: {change.Current.id}");
                        break;

                    case ChangeFeedOperationType.Delete:
                        // Access deleted document's id and partition key from metadata
                        Console.WriteLine($"Deleted Document Id: {change.Metadata.Id}");
                        
                        // PartitionKey is a dictionary: key = path name, value = partition key value
                        string categoryId = change.Metadata.PartitionKey["categoryId"]?.ToString();
                        Console.WriteLine($"Partition Key: {categoryId}");
                        
                        // Check if deleted due to TTL expiration
                        if (change.Metadata.IsTimeToLiveExpired)
                        {
                            Console.WriteLine("Deleted via TTL expiration");
                        }
                        
                        // Cleanup related data
                        await RemoveFromCacheAsync(change.Metadata.Id, categoryId);
                        break;
                }
            }
        })
    .WithInstanceName("instance1")
    .WithLeaseContainer(leaseContainer)
    .Build();

await processor.StartAsync();

Hierarchical Partition Key Example

public class UserSession
{
    public string id { get; set; }
    public string tenantId { get; set; }
    public string userId { get; set; }
    public string sessionId { get; set; }
}

// Container with HPK: ["/tenantId", "/userId", "/sessionId"]
ChangeFeedProcessor processor = container
    .GetChangeFeedProcessorBuilderWithAllVersionsAndDeletes<UserSession>(
        processorName: "sessionProcessor",
        onChangesDelegate: async (context, changes, cancellationToken) =>
        {
            foreach (ChangeFeedItem<UserSession> change in changes)
            {
                if (change.Metadata.OperationType == ChangeFeedOperationType.Delete)
                {
                    Console.WriteLine($"Deleted Session Id: {change.Metadata.Id}");
                    
                    // PartitionKey dictionary contains all hierarchy levels in order
                    string tenantId = change.Metadata.PartitionKey["tenantId"]?.ToString();
                    string userId = change.Metadata.PartitionKey["userId"]?.ToString();
                    string sessionId = change.Metadata.PartitionKey["sessionId"]?.ToString();
                    
                    Console.WriteLine($"HPK: tenant={tenantId}, user={userId}, session={sessionId}");
                    
                    await CleanupSessionAsync(tenantId, userId, sessionId);
                }
            }
        })
    .WithInstanceName("instance1")
    .WithLeaseContainer(leaseContainer)
    .Build();

await processor.StartAsync();

Cache Synchronization Example

ChangeFeedProcessor processor = container
    .GetChangeFeedProcessorBuilderWithAllVersionsAndDeletes<dynamic>(
        processorName: "cacheSync",
        onChangesDelegate: async (context, changes, cancellationToken) =>
        {
            foreach (ChangeFeedItem<dynamic> change in changes)
            {
                if (change.Metadata.OperationType == ChangeFeedOperationType.Delete)
                {
                    // Get id and partition key from metadata
                    string id = change.Metadata.Id;
                    string pk = change.Metadata.PartitionKey.Values.FirstOrDefault()?.ToString();
                    
                    // Remove from cache and search index
                    await cache.RemoveAsync(id);
                    await searchIndex.DeleteAsync(id);
                    
                    Console.WriteLine($"Removed {id} from cache and index");
                }
                else
                {
                    // Update cache and index for creates/updates
                    await cache.SetAsync(change.Current.id.ToString(), change.Current);
                    await searchIndex.IndexAsync(change.Current);
                }
            }
        })
    .WithInstanceName("cacheSync1")
    .WithLeaseContainer(leaseContainer)
    .Build();

await processor.StartAsync();

Property Details

ChangeFeedMetadata.Id

• Type: string
• Populated: Delete operations only (null for create/replace)
• Description: The document id of the deleted item
• Example: "order-12345"

ChangeFeedMetadata.PartitionKey

• Type: Dictionary<string, object>
• Populated: Delete operations only (null for create/replace)
• Description: Dictionary with partition key path(s) as keys and values as objects
• Key Format: Partition key property name without leading /
• Value Types: Can be string, number, boolean, or null

Examples:

Single Partition Key (/categoryId):

{
    "categoryId": "electronics"
}

Hierarchical Partition Key (["/tenantId", "/userId", "/sessionId"]):

{
    "tenantId": "tenant123",
    "userId": "user456", 
    "sessionId": "session789"
}

Mixed Types (["/category", "/priority", "/isActive"]):

{
    "category": "electronics",
    "priority": 1,
    "isActive": true
}

Notes

1. Only for Deletes: Id and PartitionKey are only populated when OperationType == Delete
2. No Document Body: The deleted document body is not available (Current will be null)
3. No Custom Serializer: User-provided serializers are not applied to these metadata properties
4. TTL Detection: Use IsTimeToLiveExpired to distinguish TTL deletes from explicit deletes
5. Container Setup Required: Container must have ChangeFeedPolicy.FullFidelityRetention configured

[kirankumarkolli]

Is this change validated E2E with live accounts?

Can we add E2E test (please check with @NaluTripician )

[kirankumarkolli]

Also please update the PR description with usage like how external CX will use it.

[yash2710]

Is this change validated E2E with live accounts?

Can we add E2E test (please check with @NaluTripician )

No haven't validated with live accounts. Only with emulator. Let me connect with Nalu

[yash2710]

/azp run

[azure-pipelines]

Azure Pipelines successfully started running 1 pipeline(s).

[kirankumarkolli]

Ref: #5191 (comment)