@@ -26,6 +26,8 @@ namespace DurableTask.AzureStorage.Storage
2626
2727 class Table
2828 {
29+ const int MaxParallelBatchDeletes = 10 ;
30+
2931 readonly AzureStorageClient azureStorageClient ;
3032 readonly AzureStorageOrchestrationServiceStats stats ;
3133 readonly TableServiceClient tableServiceClient ;
@@ -117,6 +119,139 @@ public async Task<TableTransactionResults> DeleteBatchAsync<T>(IEnumerable<T> en
117119 return await this . ExecuteBatchAsync ( entityBatch , item => new TableTransactionAction ( TableTransactionActionType . Delete , item ) , cancellationToken : cancellationToken ) ;
118120 }
119121
122+ /// <summary>
123+ /// Deletes entities in parallel batches of up to 100. Each batch is an atomic transaction,
124+ /// but multiple batches are submitted concurrently for improved throughput.
125+ /// Concurrency is limited to <see cref="MaxParallelBatchDeletes"/> concurrent batch transactions
126+ /// to avoid overwhelming storage and starving other operations.
127+ /// If a batch fails because an entity was already deleted (404/EntityNotFound),
128+ /// it falls back to individual deletes for that batch, skipping already-deleted entities.
129+ /// </summary>
130+ public async Task < TableTransactionResults > DeleteBatchParallelAsync < T > (
131+ IReadOnlyList < T > entityBatch ,
132+ CancellationToken cancellationToken = default ) where T : ITableEntity
133+ {
134+ if ( entityBatch . Count == 0 )
135+ {
136+ return new TableTransactionResults ( Array . Empty < Response > ( ) , TimeSpan . Zero , 0 ) ;
137+ }
138+
139+ const int batchSize = 100 ;
140+ int chunkCount = ( entityBatch . Count + batchSize - 1 ) / batchSize ;
141+ var chunks = new List < List < TableTransactionAction > > ( chunkCount ) ;
142+
143+ var currentChunk = new List < TableTransactionAction > ( batchSize ) ;
144+ foreach ( T entity in entityBatch )
145+ {
146+ currentChunk . Add ( new TableTransactionAction ( TableTransactionActionType . Delete , entity ) ) ;
147+ if ( currentChunk . Count == batchSize )
148+ {
149+ chunks . Add ( currentChunk ) ;
150+ currentChunk = new List < TableTransactionAction > ( batchSize ) ;
151+ }
152+ }
153+
154+ if ( currentChunk . Count > 0 )
155+ {
156+ chunks . Add ( currentChunk ) ;
157+ }
158+
159+ var resultsBuilder = new TableTransactionResultsBuilder ( ) ;
160+ using var semaphore = new SemaphoreSlim ( MaxParallelBatchDeletes ) ;
161+
162+ var stopwatch = Stopwatch . StartNew ( ) ;
163+ TableTransactionResults [ ] allResults = await Task . WhenAll (
164+ chunks . Select ( async chunk =>
165+ {
166+ await semaphore . WaitAsync ( cancellationToken ) ;
167+ try
168+ {
169+ return await this . ExecuteBatchWithFallbackAsync ( chunk , cancellationToken ) ;
170+ }
171+ finally
172+ {
173+ semaphore . Release ( ) ;
174+ }
175+ } ) ) ;
176+ stopwatch . Stop ( ) ;
177+
178+ foreach ( TableTransactionResults result in allResults )
179+ {
180+ resultsBuilder . Add ( result ) ;
181+ }
182+
183+ TableTransactionResults aggregatedResults = resultsBuilder . ToResults ( ) ;
184+ return new TableTransactionResults ( aggregatedResults . Responses , stopwatch . Elapsed , aggregatedResults . RequestCount ) ;
185+ }
186+
187+ /// <summary>
188+ /// Executes a batch transaction. If it fails due to an entity not found (404),
189+ /// falls back to individual delete operations, skipping entities that are already gone.
190+ /// </summary>
191+ async Task < TableTransactionResults > ExecuteBatchWithFallbackAsync (
192+ List < TableTransactionAction > batch ,
193+ CancellationToken cancellationToken )
194+ {
195+ try
196+ {
197+ return await this . ExecuteBatchAsync ( batch , cancellationToken ) ;
198+ }
199+ catch ( DurableTaskStorageException ex ) when ( ex . HttpStatusCode == 404 )
200+ {
201+ // One or more entities in the batch were already deleted.
202+ // Fall back to individual deletes, skipping 404s.
203+ // Count the failed batch attempt as 1 storage request.
204+ TableTransactionResults fallbackResults = await this . DeleteEntitiesIndividuallyAsync ( batch , cancellationToken ) ;
205+ return new TableTransactionResults (
206+ fallbackResults . Responses ,
207+ fallbackResults . Elapsed ,
208+ fallbackResults . RequestCount + 1 ) ;
209+ }
210+ catch ( RequestFailedException ex ) when ( ex . Status == 404 )
211+ {
212+ TableTransactionResults fallbackResults = await this . DeleteEntitiesIndividuallyAsync ( batch , cancellationToken ) ;
213+ return new TableTransactionResults (
214+ fallbackResults . Responses ,
215+ fallbackResults . Elapsed ,
216+ fallbackResults . RequestCount + 1 ) ;
217+ }
218+ }
219+
220+ async Task < TableTransactionResults > DeleteEntitiesIndividuallyAsync (
221+ List < TableTransactionAction > batch ,
222+ CancellationToken cancellationToken )
223+ {
224+ var responses = new List < Response > ( ) ;
225+ var stopwatch = Stopwatch . StartNew ( ) ;
226+ int requestCount = 0 ;
227+
228+ foreach ( TableTransactionAction action in batch )
229+ {
230+ requestCount ++ ;
231+ try
232+ {
233+ Response response = await this . tableClient . DeleteEntityAsync (
234+ action . Entity . PartitionKey ,
235+ action . Entity . RowKey ,
236+ action . Entity . ETag ,
237+ cancellationToken ) . DecorateFailure ( ) ;
238+ responses . Add ( response ) ;
239+ this . stats . TableEntitiesWritten . Increment ( ) ;
240+ }
241+ catch ( DurableTaskStorageException ex ) when ( ex . HttpStatusCode == 404 )
242+ {
243+ // Entity already deleted; skip.
244+ }
245+ catch ( RequestFailedException ex ) when ( ex . Status == 404 )
246+ {
247+ // Entity already deleted; skip.
248+ }
249+ }
250+
251+ stopwatch . Stop ( ) ;
252+ return new TableTransactionResults ( responses , stopwatch . Elapsed , requestCount ) ;
253+ }
254+
120255 public async Task < TableTransactionResults > InsertOrMergeBatchAsync < T > ( IEnumerable < T > entityBatch , CancellationToken cancellationToken = default ) where T : ITableEntity
121256 {
122257 TableTransactionResults results = await this . ExecuteBatchAsync ( entityBatch , item => new TableTransactionAction ( TableTransactionActionType . UpsertMerge , item ) , cancellationToken : cancellationToken ) ;
0 commit comments