[Cosmos] BUG: Count() queries do not work for cross-partition conflict feed queries

[simorenoh]

When querying the conflict feed using container.query_conflicts method doing a count query like "select value count(1) from c", partitioned queries will work and return the proper number, but cross partition ones will fail altogether. There's likely some changes we are missing in the current query pipeline for this to work with our cross partition execution.

To reproduce, you can create conflicts like this:
`
import threading
import uuid
from azure.cosmos import CosmosClient, PartitionKey, exceptions, ThroughputProperties

client = CosmosClient(COSMOS_URI, COSMOS_KEY)
database = client.create_database_if_not_exists(DATABASE_NAME)
container = database.create_container_if_not_exists(
id=CONTAINER_NAME,
offer_throughput=ThroughputProperties(auto_scale_max_throughput=1000),
partition_key=PartitionKey(path="/partitionKey"),
conflict_resolution_policy={
"mode": "custom", # Custom conflict resolution is required for conflicts to show in the conflict feed
},
)

def write_to_region(document_id, partition_key, content, region_client, iterations):
"""
Writes a document to the specified region client to create conflicts.
"""
for i in range(iterations):
print (f"Iteration: {i}; Content: {content}; WriteEndpoint: {region_client.client_connection.WriteEndpoint};")
try:
document = {
"id": document_id,
"partitionKey": partition_key,
"content": content + "-" + str(i),
}
region_client.upsert_item(document)
except exceptions.CosmosHttpResponseError as e:
print(f"Write failed: {e}")

region1_client = CosmosClient(COSMOS_URI, COSMOS_KEY, preferred_locations=["East US"]).get_database_client(DATABASE_NAME).get_container_client(CONTAINER_NAME)

for i in range(0,1):
document_id = str(uuid.uuid4()) # Shared document ID to ensure conflicts
partition_key1 = "TestPartition1"
partition_key2 = "TestPartition2"

threads = [
    threading.Thread(target=write_to_region, args=(document_id, partition_key1, "East US", region1_client, 10)),
    threading.Thread(target=write_to_region, args=(document_id, partition_key2, "East US", region1_client, 10)),
]

# Start threads
for thread in threads:
    thread.start()

# Wait for threads to complete
for thread in threads:
    thread.join()

`

[github-actions]

Thanks for the feedback! We are routing this to the appropriate team for follow-up. cc @AbhinavTrips @bambriz @Pilchie @pjohari-ms @simorenoh.