.

Author: vplauzon

code/KustoCopyConsole/Entity/InMemory/ActivityFlatHierarchy.cs

@@ -10,5 +10,7 @@ namespace KustoCopyConsole.Entity.InMemory
     internal record ActivityFlatHierarchy(
         ActivityRowItem Activity,
         IterationRowItem Iteration,
-        BlockRowItem BlockItem);
+        TempTableRowItem? TempTable,
+        BlockRowItem Block,
+        IEnumerable<UrlRowItem> Urls);
 }

code/KustoCopyConsole/Entity/InMemory/RowItemInMemoryCache.cs

@@ -42,12 +42,15 @@ public IEnumerable<ActivityFlatHierarchy> GetActivityFlatHierarchy(
                 .SelectMany(a => a.IterationMap.Values.Where(i => iterationPredicate(i)).Select(i => new
                 {
                     Activity = a,
-                    Iteration = i
+                    Iteration = i,
+                    TempTableItem = i.TempTable
                 }))
                 .SelectMany(o => o.Iteration.BlockMap.Values.Select(b => new ActivityFlatHierarchy(
                     o.Activity.RowItem,
                     o.Iteration.RowItem,
-                    b.RowItem)));
+                    o.TempTableItem,
+                    b.RowItem,
+                    b.UrlMap.Values.Select(u => u.RowItem))));
         }
 
         public IEnumerable<RowItemBase> GetItems()

code/KustoCopyConsole/Runner/AwaitExportedRunner.cs

@@ -60,12 +60,12 @@ private IEnumerable<ClusterBlocks> GetClusterBlocks()
                 a => a.RowItem.State != ActivityState.Completed,
                 i => i.RowItem.State != IterationState.Completed);
             var exportingBlocks = hierarchy
-                .Where(h => h.BlockItem.State == BlockState.Exporting);
+                .Where(h => h.Block.State == BlockState.Exporting);
             var clusterBlocks = exportingBlocks
                 .GroupBy(h => h.Activity.SourceTable.ClusterUri)
                 .Select(g => new ClusterBlocks(
                     g.Key,
-                    g.Select(h => h.BlockItem).OrderBy(b => b.Updated).Take(MAX_OPERATIONS)));
+                    g.Select(h => h.Block).OrderBy(b => b.Updated).Take(MAX_OPERATIONS)));
 
             return clusterBlocks;
         }

code/KustoCopyConsole/Runner/AwaitIngestRunner.cs

@@ -26,7 +26,7 @@ public async Task RunAsync(CancellationToken ct)
                     a => a.RowItem.State != ActivityState.Completed,
                     i => i.RowItem.State != IterationState.Completed);
                 var queuedBlocks = allBlocks
-                    .Where(h => h.BlockItem.State == BlockState.Queued);
+                    .Where(h => h.Block.State == BlockState.Queued);
                 var ingestionTasks = queuedBlocks
                     .Select(h => UpdateQueuedBlockAsync(h, ct))
                     .ToImmutableArray();
@@ -42,19 +42,16 @@ private async Task UpdateQueuedBlockAsync(
             ActivityFlatHierarchy item,
             CancellationToken ct)
         {
-            var iterationCache = RowItemGateway.InMemoryCache
-                .ActivityMap[item.Activity.ActivityName]
-                .IterationMap[item.Iteration.IterationId];
-            var tempTableName = iterationCache.TempTable!.TempTableName;
-            var targetRowCount = iterationCache.BlockMap[item.BlockItem.BlockId].UrlMap.Values
-                .Sum(u => u.RowItem.RowCount);
+            var targetRowCount = item
+                .Urls
+                .Sum(u => u.RowCount);
             var dbClient = DbClientFactory.GetDbCommandClient(
                 item.Activity.DestinationTable.ClusterUri,
                 item.Activity.DestinationTable.DatabaseName);
             var rowCount = await dbClient.GetExtentRowCountAsync(
-                new KustoPriority(item.BlockItem.GetIterationKey()),
-                tempTableName,
-                item.BlockItem.BlockTag,
+                new KustoPriority(item.Block.GetIterationKey()),
+                item.TempTable!.TempTableName,
+                item.Block.BlockTag,
                 ct);
 
             if (rowCount > targetRowCount)
@@ -65,7 +62,7 @@ private async Task UpdateQueuedBlockAsync(
             }
             if (rowCount == targetRowCount)
             {
-                var newBlockItem = item.BlockItem.ChangeState(BlockState.Ingested);
+                var newBlockItem = item.Block.ChangeState(BlockState.Ingested);
 
                 RowItemGateway.Append(newBlockItem);
             }

code/KustoCopyConsole/Runner/ExportingRunner.cs

@@ -54,21 +54,21 @@ async Task ProcessBlockAsync(
                     item.Activity.SourceTable.DatabaseName);
                 var folderPath = $"activities/{item.Activity.ActivityName}" +
                     $"iterations/{item.Iteration.IterationId:D20}" +
-                    $"/blocks/{item.BlockItem.BlockId:D20}";
+                    $"/blocks/{item.Block.BlockId:D20}";
                 var writableUris = await StagingBlobUriProvider.GetWritableFolderUrisAsync(
                     folderPath,
                     ct);
                 var query = Parameterization.Activities[item.Activity.ActivityName].KqlQuery;
                 var operationId = await dbClient.ExportBlockAsync(
-                    new KustoPriority(item.BlockItem.GetIterationKey()),
+                    new KustoPriority(item.Block.GetIterationKey()),
                     writableUris,
                     query,
                     item.Iteration.CursorStart,
                     item.Iteration.CursorEnd,
-                    item.BlockItem.IngestionTimeStart,
-                    item.BlockItem.IngestionTimeEnd,
+                    item.Block.IngestionTimeStart,
+                    item.Block.IngestionTimeEnd,
                     ct);
-                var blockItem = item.BlockItem.ChangeState(BlockState.Exporting);
+                var blockItem = item.Block.ChangeState(BlockState.Exporting);
 
                 blockItem.OperationId = operationId;
                 RowItemGateway.Append(blockItem);
@@ -96,8 +96,8 @@ private async Task<IEnumerable<ActivityFlatHierarchy>> GetExportLineUpAsync(
                 .Select(g => new
                 {
                     ClusterUri = g.Key,
-                    ExportingCount = g.Count(h => h.BlockItem.State == BlockState.Exporting),
-                    Candidates = g.Where(h => h.BlockItem.State == BlockState.Planned)
+                    ExportingCount = g.Count(h => h.Block.State == BlockState.Exporting),
+                    Candidates = g.Where(h => h.Block.State == BlockState.Planned)
                 })
                 //  Keep only clusters with candidates
                 .Where(o => o.Candidates.Any())
@@ -119,7 +119,7 @@ private async Task<IEnumerable<ActivityFlatHierarchy>> GetExportLineUpAsync(
                 .Where(o => o.ExportingAvailability > 0)
                 //  Select candidates by priority
                 .SelectMany(o => o.Candidates
-                .OrderBy(c => new KustoPriority(c.BlockItem.GetIterationKey()))
+                .OrderBy(c => new KustoPriority(c.Block.GetIterationKey()))
                 .Take(o.ExportingAvailability))
                 .ToImmutableArray();