Simplify gateway

vplauzon · vplauzon · commit ed83ccf038cd · 2025-01-15T19:09:54.000-05:00
diff --git a/code/KustoCopyConsole/Runner/AwaitIngestRunner.cs b/code/KustoCopyConsole/Runner/AwaitIngestRunner.cs
@@ -61,7 +61,7 @@ private async Task<BlockRowItem> AwaitIngestionAsync(
                 if (rowCount == targetRowCount)
                 {
                     blockItem = blockItem.ChangeState(BlockState.Ingested);
-                    await RowItemGateway.AppendAsync(blockItem, ct);
+                    RowItemGateway.Append(blockItem);
 
                     return blockItem;
                 }
diff --git a/code/KustoCopyConsole/Runner/ExportingRunner.cs b/code/KustoCopyConsole/Runner/ExportingRunner.cs
@@ -35,7 +35,7 @@ public async Task<BlockRowItem> RunAsync(
 
             if (blockItem.State == BlockState.CompletingExport)
             {
-                blockItem = await CleanUrlsAsync(blockItem, ct);
+                blockItem = CleanUrls(blockItem);
             }
             if (blockItem.State == BlockState.Planned)
             {
@@ -75,13 +75,13 @@ private async Task<BlockRowItem> AwaitExportBlockAsync(
                 throw new InvalidDataException("No URL exported");
             }
             blockItem = blockItem.ChangeState(BlockState.CompletingExport);
-            await RowItemGateway.AppendAsync(blockItem, ct);
+            RowItemGateway.Append(blockItem);
             foreach (var urlItem in urlItems)
             {
-                await RowItemGateway.AppendAsync(urlItem, ct);
+                RowItemGateway.Append(urlItem);
             }
             blockItem = blockItem.ChangeState(BlockState.Exported);
-            await RowItemGateway.AppendAsync(blockItem, ct);
+            RowItemGateway.Append(blockItem);
 
             return blockItem;
         }
@@ -112,14 +112,12 @@ private async Task<BlockRowItem> ExportBlockAsync(
 
             blockItem = blockItem.ChangeState(BlockState.Exporting);
             blockItem.OperationId = operationId;
-            await RowItemGateway.AppendAsync(blockItem, ct);
+            RowItemGateway.Append(blockItem);
 
             return blockItem;
         }
 
-        private async Task<BlockRowItem> CleanUrlsAsync(
-            BlockRowItem blockItem,
-            CancellationToken ct)
+        private BlockRowItem CleanUrls(BlockRowItem blockItem)
         {
             var existingUrls = RowItemGateway.InMemoryCache
                 .ActivityMap[blockItem.ActivityName]
@@ -130,12 +128,10 @@ private async Task<BlockRowItem> CleanUrlsAsync(
 
             foreach (var url in existingUrls)
             {
-                await RowItemGateway.AppendAsync(
-                    url.RowItem.ChangeState(UrlState.Deleted),
-                    ct);
+                RowItemGateway.Append(url.RowItem.ChangeState(UrlState.Deleted));
             }
             blockItem = blockItem.ChangeState(BlockState.Exporting);
-            await RowItemGateway.AppendAsync(blockItem, ct);
+            RowItemGateway.Append(blockItem);
 
             return blockItem;
         }
diff --git a/code/KustoCopyConsole/Runner/IterationRunner.cs b/code/KustoCopyConsole/Runner/IterationRunner.cs
@@ -138,7 +138,7 @@ private async Task<IterationRowItem> StartIterationAsync(
 
                 iterationItem = iterationItem.ChangeState(TableState.Planning);
                 iterationItem.CursorEnd = cursorEnd;
-                await RowItemGateway.AppendAsync(iterationItem, ct);
+                RowItemGateway.Append(iterationItem);
             }
 
             return iterationItem;
diff --git a/code/KustoCopyConsole/Runner/MainRunner.cs b/code/KustoCopyConsole/Runner/MainRunner.cs
@@ -91,7 +91,7 @@ private async Task RunActivityAsync(
                     $"'{activityParam.TableOption.ExportMode}' isn't supported yet");
             }
 
-            await EnsureActivityAsync(activityParam, ct);
+            EnsureActivity(activityParam);
 
             var cache = RowItemGateway.InMemoryCache;
             var activity = cache.ActivityMap[activityParam.ActivityName].RowItem;
@@ -130,7 +130,7 @@ private async Task RunActivityAsync(
                     CursorEnd = string.Empty
                 };
 
-                await RowItemGateway.AppendAsync(newIterationItem, ct);
+                RowItemGateway.Append(newIterationItem);
                 await iterationRunner.RunAsync(newIterationItem, ct);
             }
             else
@@ -140,9 +140,7 @@ await Task.WhenAll(activeIterations
             }
         }
 
-        private async Task EnsureActivityAsync(
-            ActivityParameterization activityParam,
-            CancellationToken ct)
+        private void EnsureActivity(ActivityParameterization activityParam)
         {
             if (!RowItemGateway.InMemoryCache.ActivityMap.ContainsKey(activityParam.ActivityName))
             {
@@ -154,7 +152,7 @@ private async Task EnsureActivityAsync(
                     DestinationTable = activityParam.GetEffectiveDestinationTableIdentity()
                 };
 
-                await RowItemGateway.AppendAsync(activity, ct);
+                RowItemGateway.Append(activity);
             }
         }
     }
diff --git a/code/KustoCopyConsole/Runner/MoveExtentsRunner.cs b/code/KustoCopyConsole/Runner/MoveExtentsRunner.cs
@@ -55,7 +55,7 @@ private async Task<BlockRowItem> MoveExtentsAsync(
                     ct);
 
                 blockItem = blockItem.ChangeState(BlockState.ExtentMoved);
-                await RowItemGateway.AppendAsync(blockItem, ct);
+                RowItemGateway.Append(blockItem);
             }
 
             return blockItem;
diff --git a/code/KustoCopyConsole/Runner/PlanningRunner.cs b/code/KustoCopyConsole/Runner/PlanningRunner.cs
@@ -95,14 +95,14 @@ private async Task<IterationRowItem> PlanBlocksAsync(
 
                         orderedDistributionInExtents = remainingDistributionInExtents
                             .ToImmutableArray();
-                        await RowItemGateway.AppendAsync(newBlockItem, ct);
+                        RowItemGateway.Append(newBlockItem);
                         lastBlock = newBlockItem;
                     }
                 }
                 else
                 {
                     iterationItem = iterationItem.ChangeState(TableState.Planned);
-                    await RowItemGateway.AppendAsync(iterationItem, ct);
+                    RowItemGateway.Append(iterationItem);
                 }
             }
 
diff --git a/code/KustoCopyConsole/Runner/QueueIngestRunner.cs b/code/KustoCopyConsole/Runner/QueueIngestRunner.cs
@@ -63,7 +63,7 @@ private async Task<BlockRowItem> QueueIngestBlockAsync(
 
             blockItem = blockItem.ChangeState(BlockState.Queued);
             blockItem.BlockTag = blockTag;
-            await RowItemGateway.AppendAsync(blockItem, ct);
+            RowItemGateway.Append(blockItem);
 
             return blockItem;
         }
diff --git a/code/KustoCopyConsole/Runner/TempTableCreatingRunner.cs b/code/KustoCopyConsole/Runner/TempTableCreatingRunner.cs
@@ -62,11 +62,7 @@ await dbCommandClient.CreateTempTableAsync(
                 ct);
 
             iterationItem = iterationItem.ChangeState(TableState.TempTableCreated);
-
-            var rowItemAppend = await RowItemGateway.AppendAsync(iterationItem, ct);
-
-            //  We want to make sure this is recorded before we start ingesting data into it
-            await rowItemAppend.ItemAppendTask;
+            RowItemGateway.Append(iterationItem);
 
             return iterationItem;
         }
@@ -84,11 +80,9 @@ private async Task<IterationRowItem> PrepareTempTableAsync(
             iterationItem = iterationItem.ChangeState(TableState.TempTableCreating);
             iterationItem.TempTableName = tempTableName;
 
-            var rowItemAppend = await RowItemGateway.AppendAsync(iterationItem, ct);
-
             //  We want to ensure the item is appended before creating a temp table so
             //  we don't lose track of the table
-            await rowItemAppend.ItemAppendTask;
+            await RowItemGateway.AppendAndPersistAsync(iterationItem, ct);
 
             return iterationItem;
         }
diff --git a/code/KustoCopyConsole/Storage/RowItemAppend.cs b/code/KustoCopyConsole/Storage/RowItemAppend.cs
@@ -7,5 +7,5 @@
 
 namespace KustoCopyConsole.Storage
 {
-    internal record RowItemAppend(RowItemBase Item, Task ItemAppendTask);
+    internal record RowItemAppend(RowItemBase Item);
 }
diff --git a/code/KustoCopyConsole/Storage/RowItemGateway.cs b/code/KustoCopyConsole/Storage/RowItemGateway.cs

Original file line number	Diff line number	Diff line change
`@@ -61,7 +61,7 @@ private async Task<BlockRowItem> AwaitIngestionAsync(`
`61`	`61`	`if (rowCount == targetRowCount)`
`62`	`62`	`{`
`63`	`63`	`blockItem = blockItem.ChangeState(BlockState.Ingested);`
`64`		`- await RowItemGateway.AppendAsync(blockItem, ct);`
	`64`	`+ RowItemGateway.Append(blockItem);`
`65`	`65`
`66`	`66`	`return blockItem;`
`67`	`67`	`}`
Original file line number	Diff line number	Diff line change
`@@ -35,7 +35,7 @@ public async Task<BlockRowItem> RunAsync(`
`35`	`35`
`36`	`36`	`if (blockItem.State == BlockState.CompletingExport)`
`37`	`37`	`{`
`38`		`- blockItem = await CleanUrlsAsync(blockItem, ct);`
	`38`	`+ blockItem = CleanUrls(blockItem);`
`39`	`39`	`}`
`40`	`40`	`if (blockItem.State == BlockState.Planned)`
`41`	`41`	`{`
`@@ -75,13 +75,13 @@ private async Task<BlockRowItem> AwaitExportBlockAsync(`
`75`	`75`	`throw new InvalidDataException("No URL exported");`
`76`	`76`	`}`
`77`	`77`	`blockItem = blockItem.ChangeState(BlockState.CompletingExport);`
`78`		`- await RowItemGateway.AppendAsync(blockItem, ct);`
	`78`	`+ RowItemGateway.Append(blockItem);`
`79`	`79`	`foreach (var urlItem in urlItems)`
`80`	`80`	`{`
`81`		`- await RowItemGateway.AppendAsync(urlItem, ct);`
	`81`	`+ RowItemGateway.Append(urlItem);`
`82`	`82`	`}`
`83`	`83`	`blockItem = blockItem.ChangeState(BlockState.Exported);`
`84`		`- await RowItemGateway.AppendAsync(blockItem, ct);`
	`84`	`+ RowItemGateway.Append(blockItem);`
`85`	`85`
`86`	`86`	`return blockItem;`
`87`	`87`	`}`
`@@ -112,14 +112,12 @@ private async Task<BlockRowItem> ExportBlockAsync(`
`112`	`112`
`113`	`113`	`blockItem = blockItem.ChangeState(BlockState.Exporting);`
`114`	`114`	`blockItem.OperationId = operationId;`
`115`		`- await RowItemGateway.AppendAsync(blockItem, ct);`
	`115`	`+ RowItemGateway.Append(blockItem);`
`116`	`116`
`117`	`117`	`return blockItem;`
`118`	`118`	`}`
`119`	`119`
`120`		`- private async Task<BlockRowItem> CleanUrlsAsync(`
`121`		`- BlockRowItem blockItem,`
`122`		`- CancellationToken ct)`
	`120`	`+ private BlockRowItem CleanUrls(BlockRowItem blockItem)`
`123`	`121`	`{`
`124`	`122`	`var existingUrls = RowItemGateway.InMemoryCache`
`125`	`123`	`.ActivityMap[blockItem.ActivityName]`
`@@ -130,12 +128,10 @@ private async Task<BlockRowItem> CleanUrlsAsync(`
`130`	`128`
`131`	`129`	`foreach (var url in existingUrls)`
`132`	`130`	`{`
`133`		`- await RowItemGateway.AppendAsync(`
`134`		`- url.RowItem.ChangeState(UrlState.Deleted),`
`135`		`- ct);`
	`131`	`+ RowItemGateway.Append(url.RowItem.ChangeState(UrlState.Deleted));`
`136`	`132`	`}`
`137`	`133`	`blockItem = blockItem.ChangeState(BlockState.Exporting);`
`138`		`- await RowItemGateway.AppendAsync(blockItem, ct);`
	`134`	`+ RowItemGateway.Append(blockItem);`
`139`	`135`
`140`	`136`	`return blockItem;`
`141`	`137`	`}`
Original file line number	Diff line number	Diff line change
`@@ -138,7 +138,7 @@ private async Task<IterationRowItem> StartIterationAsync(`
`138`	`138`
`139`	`139`	`iterationItem = iterationItem.ChangeState(TableState.Planning);`
`140`	`140`	`iterationItem.CursorEnd = cursorEnd;`
`141`		`- await RowItemGateway.AppendAsync(iterationItem, ct);`
	`141`	`+ RowItemGateway.Append(iterationItem);`
`142`	`142`	`}`
`143`	`143`
`144`	`144`	`return iterationItem;`
Original file line number	Diff line number	Diff line change
`@@ -91,7 +91,7 @@ private async Task RunActivityAsync(`
`91`	`91`	`$"'{activityParam.TableOption.ExportMode}' isn't supported yet");`
`92`	`92`	`}`
`93`	`93`
`94`		`- await EnsureActivityAsync(activityParam, ct);`
	`94`	`+ EnsureActivity(activityParam);`
`95`	`95`
`96`	`96`	`var cache = RowItemGateway.InMemoryCache;`
`97`	`97`	`var activity = cache.ActivityMap[activityParam.ActivityName].RowItem;`
`@@ -130,7 +130,7 @@ private async Task RunActivityAsync(`
`130`	`130`	`CursorEnd = string.Empty`
`131`	`131`	`};`
`132`	`132`
`133`		`- await RowItemGateway.AppendAsync(newIterationItem, ct);`
	`133`	`+ RowItemGateway.Append(newIterationItem);`
`134`	`134`	`await iterationRunner.RunAsync(newIterationItem, ct);`
`135`	`135`	`}`
`136`	`136`	`else`
`@@ -140,9 +140,7 @@ await Task.WhenAll(activeIterations`
`140`	`140`	`}`
`141`	`141`	`}`
`142`	`142`
`143`		`- private async Task EnsureActivityAsync(`
`144`		`- ActivityParameterization activityParam,`
`145`		`- CancellationToken ct)`
	`143`	`+ private void EnsureActivity(ActivityParameterization activityParam)`
`146`	`144`	`{`
`147`	`145`	`if (!RowItemGateway.InMemoryCache.ActivityMap.ContainsKey(activityParam.ActivityName))`
`148`	`146`	`{`
`@@ -154,7 +152,7 @@ private async Task EnsureActivityAsync(`
`154`	`152`	`DestinationTable = activityParam.GetEffectiveDestinationTableIdentity()`
`155`	`153`	`};`
`156`	`154`
`157`		`- await RowItemGateway.AppendAsync(activity, ct);`
	`155`	`+ RowItemGateway.Append(activity);`
`158`	`156`	`}`
`159`	`157`	`}`
`160`	`158`	`}`
Original file line number	Diff line number	Diff line change
`@@ -55,7 +55,7 @@ private async Task<BlockRowItem> MoveExtentsAsync(`
`55`	`55`	`ct);`
`56`	`56`
`57`	`57`	`blockItem = blockItem.ChangeState(BlockState.ExtentMoved);`
`58`		`- await RowItemGateway.AppendAsync(blockItem, ct);`
	`58`	`+ RowItemGateway.Append(blockItem);`
`59`	`59`	`}`
`60`	`60`
`61`	`61`	`return blockItem;`
Original file line number	Diff line number	Diff line change
`@@ -95,14 +95,14 @@ private async Task<IterationRowItem> PlanBlocksAsync(`
`95`	`95`
`96`	`96`	`orderedDistributionInExtents = remainingDistributionInExtents`
`97`	`97`	`.ToImmutableArray();`
`98`		`- await RowItemGateway.AppendAsync(newBlockItem, ct);`
	`98`	`+ RowItemGateway.Append(newBlockItem);`
`99`	`99`	`lastBlock = newBlockItem;`
`100`	`100`	`}`
`101`	`101`	`}`
`102`	`102`	`else`
`103`	`103`	`{`
`104`	`104`	`iterationItem = iterationItem.ChangeState(TableState.Planned);`
`105`		`- await RowItemGateway.AppendAsync(iterationItem, ct);`
	`105`	`+ RowItemGateway.Append(iterationItem);`
`106`	`106`	`}`
`107`	`107`	`}`
`108`	`108`
Original file line number	Diff line number	Diff line change
`@@ -63,7 +63,7 @@ private async Task<BlockRowItem> QueueIngestBlockAsync(`
`63`	`63`
`64`	`64`	`blockItem = blockItem.ChangeState(BlockState.Queued);`
`65`	`65`	`blockItem.BlockTag = blockTag;`
`66`		`- await RowItemGateway.AppendAsync(blockItem, ct);`
	`66`	`+ RowItemGateway.Append(blockItem);`
`67`	`67`
`68`	`68`	`return blockItem;`
`69`	`69`	`}`
Original file line number	Diff line number	Diff line change
`@@ -7,5 +7,5 @@`
`7`	`7`
`8`	`8`	`namespace KustoCopyConsole.Storage`
`9`	`9`	`{`
`10`		`- internal record RowItemAppend(RowItemBase Item, Task ItemAppendTask);`
	`10`	`+ internal record RowItemAppend(RowItemBase Item);`
`11`	`11`	`}`