Skip to content

Update Main Branch Changes with Azure Storage Track2 Methods #1081

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

Merged
nytian merged 1 commit into from
May 7, 2024
Merged

Update Main Branch Changes with Azure Storage Track2 Methods #1081

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
17 changes: 12 additions & 5 deletions src/DurableTask.AzureStorage/EntityTrackingStoreQueries.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ namespace DurableTask.AzureStorage
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Azure;
using DurableTask.AzureStorage.Tracking;
using DurableTask.Core;
using DurableTask.Core.Entities;
Expand Down Expand Up @@ -56,7 +57,7 @@ public EntityTrackingStoreQueries(
CancellationToken cancellation = default(CancellationToken))
{
await this.ensureTaskHub();
OrchestrationState? state = (await this.trackingStore.GetStateAsync(id.ToString(), allExecutions: false, fetchInput: includeState)).FirstOrDefault();
OrchestrationState? state = await this.trackingStore.GetStateAsync(id.ToString(), allExecutions: false, fetchInput: includeState).FirstOrDefaultAsync();
return await this.GetEntityMetadataAsync(state, includeStateless, includeState);
}

Expand Down Expand Up @@ -87,7 +88,10 @@ public async override Task<EntityQueryResult> QueryEntitiesAsync(EntityQuery fil

do
{
DurableStatusQueryResult result = await this.trackingStore.GetStateAsync(condition, filter.PageSize ?? 100, continuationToken, cancellation);
Page<OrchestrationState>? page = await this.trackingStore.GetStateAsync(condition, cancellation).AsPages(continuationToken, filter.PageSize ?? 100).FirstOrDefaultAsync();
DurableStatusQueryResult result = page != null
? new DurableStatusQueryResult { ContinuationToken = page.ContinuationToken, OrchestrationState = page.Values }
: new DurableStatusQueryResult { OrchestrationState = Array.Empty<OrchestrationState>() };
entityResult = await ConvertResultsAsync(result.OrchestrationState);
continuationToken = result.ContinuationToken;
}
Expand Down Expand Up @@ -139,7 +143,10 @@ async ValueTask<List<EntityMetadata>> ConvertResultsAsync(IEnumerable<Orchestrat
// perform that action. Waits for all actions to finish after each page.
do
{
DurableStatusQueryResult page = await this.trackingStore.GetStateAsync(condition, 100, continuationToken, cancellation);
Page<OrchestrationState>? states = await this.trackingStore.GetStateAsync(condition, cancellation).AsPages(continuationToken, 100).FirstOrDefaultAsync();
DurableStatusQueryResult page = states != null
? new DurableStatusQueryResult { ContinuationToken = states.ContinuationToken, OrchestrationState = states.Values }
: new DurableStatusQueryResult { OrchestrationState = Array.Empty<OrchestrationState>() };
continuationToken = page.ContinuationToken;

var tasks = new List<Task>();
Expand Down Expand Up @@ -174,8 +181,8 @@ async Task DeleteIdleOrchestrationEntity(OrchestrationState state)

async Task CheckForOrphanedLockAndFixIt(OrchestrationState state, string lockOwner)
{
OrchestrationState? ownerState
= (await this.trackingStore.GetStateAsync(lockOwner, allExecutions: false, fetchInput: false)).FirstOrDefault();
OrchestrationState? ownerState
= await this.trackingStore.GetStateAsync(lockOwner, allExecutions: false, fetchInput: false).FirstOrDefaultAsync();

bool OrchestrationIsRunning(OrchestrationStatus? status)
=> status != null && (status == OrchestrationStatus.Running || status == OrchestrationStatus.Suspended);
Expand Down
5 changes: 1 addition & 4 deletions src/DurableTask.AzureStorage/Tracking/OrchestrationInstanceStatusQueryCondition.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -145,10 +145,7 @@ internal ODataCondition ToOData()
}
else if (this.ExcludeEntities)
{
conditions.Add(TableQuery.CombineFilters(
TableQuery.GenerateFilterCondition("PartitionKey", QueryComparisons.LessThan, "@"),
TableOperators.Or,
TableQuery.GenerateFilterCondition("PartitionKey", QueryComparisons.GreaterThanOrEqual, "A")));
conditions.Add($"{nameof(OrchestrationInstanceStatus.PartitionKey)} lt '@' or {nameof(OrchestrationInstanceStatus.PartitionKey)} ge 'A'");
Copy link
Collaborator Author

@nytian nytian May 1, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In Azure Storage Track2, we don't have TableQuery anymore, so I translate the original query with this to align with other context.

Copy link
Collaborator

@davidmrdavid davidmrdavid May 3, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can you please add a comment on top of this method that documents the query? It's a big hard to follow as-is

Copy link
Collaborator Author

@nytian nytian May 3, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just want to confirm you mean what $"{nameof(OrchestrationInstanceStatus.PartitionKey)} lt '@' or {nameof(OrchestrationInstanceStatus.PartitionKey)} ge 'A'") means, right?

}

if (this.InstanceId != null)
Expand Down