Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

ReadMany: Fixes ReadMany API for None Partition key values #2661

Merged
merged 10 commits into from
Aug 20, 2021
Merged

ReadMany: Fixes ReadMany API for None Partition key values #2661

Show file tree
Hide file tree
Changes from 7 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
a4088ca
Fixing ReadMany API for None Partition key values
asketagarwal Aug 9, 2021
48b22eb
addressing comments
asketagarwal Aug 10, 2021
66495ab
fixing build
asketagarwal Aug 10, 2021
ec52b38
fixing build
asketagarwal Aug 11, 2021
026f397
Merge branch 'master' into users/askagarw/readmanypknonefix
j82w Aug 13, 2021
d89882b
Fixing none pk values query creation
asketagarwal Aug 18, 2021
2fbc11c
Merge branch 'users/askagarw/readmanypknonefix' of https://github.com…
asketagarwal Aug 18, 2021
a4ad806
fixing tests
asketagarwal Aug 19, 2021
77f4995
missing return statement
asketagarwal Aug 19, 2021
ab231a7
Merge branch 'master' into users/askagarw/readmanypknonefix
j82w Aug 19, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
88 changes: 62 additions & 26 deletions Microsoft.Azure.Cosmos/src/ReadManyQueryHelper.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,14 +42,22 @@ public override async Task<ResponseMessage> ExecuteReadManyRequestAsync(IReadOnl
ITrace trace,
CancellationToken cancellationToken)
{
string resourceId = await this.container.GetCachedRIDAsync(cancellationToken);
IDictionary<PartitionKeyRange, List<(string, PartitionKey)>> partitionKeyRangeItemMap =
await this.CreatePartitionKeyRangeItemListMapAsync(items, cancellationToken);
IDictionary<PartitionKeyRange, List<(string, PartitionKey)>> partitionKeyRangeItemMap;
string resourceId;
try
{
resourceId = await this.container.GetCachedRIDAsync(cancellationToken);
partitionKeyRangeItemMap = await this.CreatePartitionKeyRangeItemListMapAsync(items, trace, cancellationToken);
}
catch (CosmosException ex)
{
return ex.ToCosmosResponseMessage(request: null);
ealsur marked this conversation as resolved.
Show resolved Hide resolved
}

List<ResponseMessage>[] queryResponses = await this.ReadManyTaskHelperAsync(partitionKeyRangeItemMap,
readManyRequestOptions,
trace,
cancellationToken);
readManyRequestOptions,
trace,
cancellationToken);

return this.CombineStreamsFromQueryResponses(queryResponses, resourceId, trace); // also disposes the response messages
}
Expand All @@ -60,7 +68,7 @@ public override async Task<FeedResponse<T>> ExecuteReadManyRequestAsync<T>(IRead
CancellationToken cancellationToken)
{
IDictionary<PartitionKeyRange, List<(string, PartitionKey)>> partitionKeyRangeItemMap =
await this.CreatePartitionKeyRangeItemListMapAsync(items, cancellationToken);
await this.CreatePartitionKeyRangeItemListMapAsync(items, trace, cancellationToken);

List<ResponseMessage>[] queryResponses = await this.ReadManyTaskHelperAsync(partitionKeyRangeItemMap,
readManyRequestOptions,
Expand Down Expand Up @@ -116,6 +124,7 @@ internal async Task<List<ResponseMessage>[]> ReadManyTaskHelperAsync(IDictionary

private async Task<IDictionary<PartitionKeyRange, List<(string, PartitionKey)>>> CreatePartitionKeyRangeItemListMapAsync(
IReadOnlyList<(string, PartitionKey)> items,
ITrace trace,
CancellationToken cancellationToken = default)
{
CollectionRoutingMap collectionRoutingMap = await this.container.GetRoutingMapAsync(cancellationToken);
Expand All @@ -125,7 +134,9 @@ internal async Task<List<ResponseMessage>[]> ReadManyTaskHelperAsync(IDictionary

foreach ((string id, PartitionKey pk) item in items)
{
string effectivePartitionKeyValue = item.pk.InternalKey.GetEffectivePartitionKeyString(this.partitionKeyDefinition);
Documents.Routing.PartitionKeyInternal partitionKeyInternal =
await this.GetPartitionKeyInternalAsync(item.pk, trace, cancellationToken);
string effectivePartitionKeyValue = partitionKeyInternal.GetEffectivePartitionKeyString(this.partitionKeyDefinition);
PartitionKeyRange partitionKeyRange = collectionRoutingMap.GetRangeByEffectivePartitionKey(effectivePartitionKeyValue);
if (partitionKeyRangeItemMap.TryGetValue(partitionKeyRange, out List<(string, PartitionKey)> itemList))
{
Expand Down Expand Up @@ -259,33 +270,45 @@ private QueryDefinition CreateReadManyQueryDefinitionForOther(List<(string, Part

queryStringBuilder.Append("SELECT * FROM c WHERE ( ");
for (int i = startIndex; i < totalItemCount; i++)
{
object[] pkValues = items[i].Item2.InternalKey.ToObjectArray();

if (pkValues.Length != this.partitionKeyDefinition.Paths.Count)
{
throw new ArgumentException("Number of components in the partition key value does not match the definition.");
}

{
string pkParamName = "@param_pk" + i;
string idParamName = "@param_id" + i;
sqlParameters.Add(new SqlParameter(idParamName, items[i].Item1));

queryStringBuilder.Append("( ");
queryStringBuilder.Append("c.id = ");
queryStringBuilder.Append(idParamName);
for (int j = 0; j < this.partitionKeySelectors.Count; j++)
if (items[i].Item2.IsNone)
{
for (int j = 0; j < this.partitionKeySelectors.Count; j++)
asketagarwal marked this conversation as resolved.
Show resolved Hide resolved
{
queryStringBuilder.Append(" AND ");
queryStringBuilder.Append("IS_DEFINED(c");
queryStringBuilder.Append(this.partitionKeySelectors[j]);
j82w marked this conversation as resolved.
Show resolved Hide resolved
queryStringBuilder.Append(") = false");
}
}
else
{
queryStringBuilder.Append(" AND ");
queryStringBuilder.Append("c");
queryStringBuilder.Append(this.partitionKeySelectors[j]);
queryStringBuilder.Append(" = ");

string pkParamNameForSinglePath = pkParamName + j;
sqlParameters.Add(new SqlParameter(pkParamNameForSinglePath, pkValues[j]));
queryStringBuilder.Append(pkParamNameForSinglePath);
object[] pkValues = items[i].Item2.InternalKey.ToObjectArray();
if (pkValues.Length != this.partitionKeyDefinition.Paths.Count)
{
throw new ArgumentException("Number of components in the partition key " +
"value does not match the definition.");
}
for (int j = 0; j < this.partitionKeySelectors.Count; j++)
{
queryStringBuilder.Append(" AND ");
queryStringBuilder.Append("c");
queryStringBuilder.Append(this.partitionKeySelectors[j]);
queryStringBuilder.Append(" = ");

string pkParamNameForSinglePath = pkParamName + j;
sqlParameters.Add(new SqlParameter(pkParamNameForSinglePath, pkValues[j]));
queryStringBuilder.Append(pkParamNameForSinglePath);
}
}

queryStringBuilder.Append(" )");

if (i < totalItemCount - 1)
Expand Down Expand Up @@ -365,6 +388,19 @@ private void CancelCancellationToken(CancellationToken cancellationToken)
}
}

private ValueTask<Documents.Routing.PartitionKeyInternal> GetPartitionKeyInternalAsync(PartitionKey partitionKey,
ITrace trace,
CancellationToken cancellationToken)
{
if (partitionKey.IsNone)
{
return new ValueTask<Documents.Routing.PartitionKeyInternal>(
this.container.GetNonePartitionKeyValueAsync(trace, cancellationToken));
}
asketagarwal marked this conversation as resolved.
Show resolved Hide resolved

return new ValueTask<Documents.Routing.PartitionKeyInternal>(partitionKey.InternalKey);
}

private class ReadManyFeedResponseEnumerable<T> : IEnumerable<T>
{
private readonly List<FeedResponse<T>> typedResponses;
Expand Down
12 changes: 11 additions & 1 deletion Microsoft.Azure.Cosmos/src/Resource/Container/ContainerCore.Items.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -298,7 +298,17 @@ public async Task<ResponseMessage> ReadManyItemsStreamAsync(
throw new ArgumentNullException(nameof(trace));
}

ReadManyHelper readManyHelper = new ReadManyQueryHelper(await this.GetPartitionKeyDefinitionAsync(),
PartitionKeyDefinition partitionKeyDefinition = null;
try
{
partitionKeyDefinition = await this.GetPartitionKeyDefinitionAsync();
}
catch (CosmosException ex)
{
ex.ToCosmosResponseMessage(request: null);
asketagarwal marked this conversation as resolved.
Show resolved Hide resolved
}

ReadManyHelper readManyHelper = new ReadManyQueryHelper(partitionKeyDefinition,
this);

return await readManyHelper.ExecuteReadManyRequestAsync(items,
Expand Down
90 changes: 90 additions & 0 deletions ...osoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.EmulatorTests/CosmosReadManyItemsTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -391,6 +391,69 @@ await container.CreateItemAsync(
await database.DeleteAsync();
}

[TestMethod]
public async Task ReadManyWithNonePkValues()
{
for (int i = 0; i < 5; i++)
{
await this.Container.CreateItemAsync(new ActivityWithNoPk("id" + i.ToString()),
PartitionKey.None);
}

List<(string, PartitionKey)> itemList = new List<(string, PartitionKey)>();
for (int i = 0; i < 5; i++)
{
itemList.Add(("id" + i.ToString(), PartitionKey.None));
}

FeedResponse<ActivityWithNoPk> feedResponse = await this.Container.ReadManyItemsAsync<ActivityWithNoPk>(itemList);
Assert.AreEqual(feedResponse.Count, 5);
asketagarwal marked this conversation as resolved.
Show resolved Hide resolved
}

[TestMethod]
public async Task ReadManyItemsFromNonPartitionedContainers()
{
ContainerInternal container = await NonPartitionedContainerHelper.CreateNonPartitionedContainer(this.database,
Guid.NewGuid().ToString());
for (int i = 0; i < 5; i++)
{
await NonPartitionedContainerHelper.CreateItemInNonPartitionedContainer(container, "id" + i.ToString());
}

// read using PartitionKey.None pk value
List<(string, PartitionKey)> itemList = new List<(string, PartitionKey)>();
for (int i = 0; i < 10; i++)
{
itemList.Add(("id" + i.ToString(), PartitionKey.None));
}

FeedResponse<ActivityWithNoPk> feedResponse = await container.ReadManyItemsAsync<ActivityWithNoPk>(itemList);
Assert.AreEqual(feedResponse.Count, 5);

// Start inserting documents with same id but new pk values
for (int i = 0; i < 5; i++)
{
await container.CreateItemAsync(new ActivityWithSystemPk("id" + i.ToString(), "newPK"),
new PartitionKey("newPK"));
}

FeedResponse<ActivityWithSystemPk> feedResponseWithPK = await container.ReadManyItemsAsync<ActivityWithSystemPk>(itemList);
Assert.AreEqual(feedResponseWithPK.Count, 5);

for (int i = 0; i < 5; i++)
{
itemList.Add(("id" + i.ToString(), new PartitionKey("newPK")));
}
feedResponseWithPK = await container.ReadManyItemsAsync<ActivityWithSystemPk>(itemList);
Assert.AreEqual(feedResponseWithPK.Count, 10);

// Mixing both None and non-None pk values
itemList = new List<(string, PartitionKey)>
{ ("id0", PartitionKey.None), ("id0", new PartitionKey("newPK")) };
feedResponse = await container.ReadManyItemsAsync<ActivityWithNoPk>(itemList);
Assert.AreEqual(feedResponse.Count, 2);
}

[TestMethod]
[DataRow(HttpStatusCode.NotFound)]
public async Task ReadManyExceptionsTest(HttpStatusCode statusCode)
Expand Down Expand Up @@ -508,5 +571,32 @@ public override async Task<ResponseMessage> SendAsync(RequestMessage requestMess
return await base.SendAsync(requestMessage, cancellationToken);
}
}

private class ActivityWithNoPk
{
public ActivityWithNoPk(string id)
{
this.id = id;
}

#pragma warning disable IDE1006 // Naming Styles
public string id { get; set; }
#pragma warning restore IDE1006 // Naming Styles
}

private class ActivityWithSystemPk
{
public ActivityWithSystemPk(string id, string pk)
{
this.id = id;
this._partitionKey = pk;
}

#pragma warning disable IDE1006 // Naming Styles
public string id { get; set; }
#pragma warning restore IDE1006 // Naming Styles

public string _partitionKey { get; set; }
}
}
}