.

Author: vplauzon

code/KustoCopyConsole/Entity/RowItems/ActivityRowItem.cs

@@ -23,7 +23,7 @@ public override void Validate()
             {
                 throw new InvalidDataException($"{nameof(ActivityName)} must have a value");
             }
-            SourceTable.Validate();
+            SourceDatabase.Validate();
             DestinationTable.Validate();
         }
 

code/KustoCopyConsole/Runner/ExportingRunner.cs

@@ -26,12 +26,10 @@ public async Task<BlockRowItem> RunAsync(
             var activityItem = RowItemGateway.InMemoryCache
                 .ActivityMap[blockItem.ActivityName]
                 .RowItem;
-            var activityParam = Parameterization.Activities[blockItem.ActivityName];
             var exportClient = DbClientFactory.GetExportClient(
-                activityItem.SourceTable.ClusterUri,
-                activityItem.SourceTable.DatabaseName,
-                activityItem.SourceTable.TableName,
-                activityParam.KqlQuery);
+                activityItem.SourceDatabase.ClusterUri,
+                blockItem.SourceTable.DatabaseName,
+                blockItem.SourceTable.TableName);
 
             if (blockItem.State == BlockState.CompletingExport)
             {
@@ -55,8 +53,8 @@ private async Task<BlockRowItem> AwaitExportBlockAsync(
             CancellationToken ct)
         {
             var exportDetails = await exportClient.AwaitExportAsync(
-                new KustoPriority(
-                    blockItem.ActivityName, blockItem.IterationId, blockItem.BlockId),
+                blockItem.IterationId,
+                blockItem.SourceTable.TableName,
                 blockItem.OperationId,
                 ct);
             var urlItems = exportDetails

code/KustoCopyConsole/Runner/IterationRunner.cs

@@ -130,8 +130,8 @@ private async Task<IterationRowItem> StartIterationAsync(
                     .ActivityMap[iterationItem.ActivityName]
                     .RowItem;
                 var queryClient = DbClientFactory.GetDbQueryClient(
-                        activity.SourceTable.ClusterUri,
-                        activity.SourceTable.DatabaseName);
+                        activity.SourceDatabase.ClusterUri,
+                        activity.SourceDatabase.DatabaseName);
                 var cursorEnd = await queryClient.GetCurrentCursorAsync(
                     new KustoPriority(iterationItem.ActivityName, iterationItem.IterationId),
                     ct);
@@ -174,7 +174,7 @@ private async Task<IterationRowItem> CreateTempTableAsync(
         #region Block Level
         private async Task ProcessAllBlocksAsync(IterationRowItem iterationItem, CancellationToken ct)
         {
-            var blobPathProvider = GetBlobPathFactory();
+            var blobPathProvider = GetBlobPathFactory(iterationItem.SourceTable);
             var blockItems = RowItemGateway.InMemoryCache
                 .ActivityMap[iterationItem.ActivityName]
                 .IterationMap[iterationItem.IterationId]
@@ -192,15 +192,28 @@ private async Task ProcessAllBlocksAsync(IterationRowItem iterationItem, Cancell
             await Task.WhenAll(processBlockTasks);
         }
 
-        private IStagingBlobUriProvider GetBlobPathFactory()
+        private IStagingBlobUriProvider GetBlobPathFactory(TableIdentity sourceTable)
         {
-            var tempUriProvider = new AzureBlobUriProvider(
-                Parameterization.StagingStorageContainers
-                .Select(s => new Uri(s))
-                .ToImmutableArray(),
-                Parameterization.GetCredentials());
+            var activity = Parameterization.Activities
+                .Values
+                .Where(a => a.Source.GetTableIdentity() == sourceTable)
+                .FirstOrDefault();
 
-            return tempUriProvider;
+            if (activity == null)
+            {
+                throw new InvalidDataException($"Can't find table in parameters:  {sourceTable}");
+            }
+            else
+            {
+                var destinationTable = activity.Destination.GetTableIdentity();
+                var tempUriProvider = new AzureBlobUriProvider(
+                    Parameterization.StagingStorageContainers
+                    .Select(s => new Uri(s))
+                    .ToImmutableArray(),
+                    Parameterization.GetCredentials());
+
+                return tempUriProvider;
+            }
         }
 
         private async Task ProcessSingleBlockAsync(

code/KustoCopyConsole/Runner/MainRunner.cs

@@ -150,7 +150,7 @@ private async Task EnsureActivityAsync(
                 {
                     State = ActivityState.Active,
                     ActivityName = activityParam.ActivityName,
-                    SourceTable = activityParam.Source.GetTableIdentity(),
+                    SourceDatabase = activityParam.Source.GetTableIdentity(),
                     DestinationTable = activityParam.GetEffectiveDestinationTableIdentity()
                 };
 

code/KustoCopyConsole/Runner/PlanningRunner.cs

@@ -56,11 +56,11 @@ private async Task<IterationRowItem> PlanBlocksAsync(
                 .ActivityMap[iterationItem.ActivityName]
                 .RowItem;
             var queryClient = DbClientFactory.GetDbQueryClient(
-                activity.SourceTable.ClusterUri,
-                activity.SourceTable.DatabaseName);
+                activity.SourceDatabase.ClusterUri,
+                activity.SourceDatabase.DatabaseName);
             var dbCommandClient = DbClientFactory.GetDbCommandClient(
-                activity.SourceTable.ClusterUri,
-                activity.SourceTable.DatabaseName);
+                activity.SourceDatabase.ClusterUri,
+                activity.SourceDatabase.DatabaseName);
 
             //  Loop on block batches
             while (iterationItem.State == TableState.Planning)
@@ -147,23 +147,18 @@ private async Task<IterationRowItem> PlanBlocksAsync(
         }
 
         //  Merge results from query + show extents command
-        private async Task<IImmutableList<RecordDistributionInExtent>> GetRecordDistributionInExtents(
-            IterationRowItem iterationItem,
+        private static async Task<IImmutableList<RecordDistributionInExtent>> GetRecordDistributionInExtents(
+            IterationRowItem sourceTableItem,
             DateTime? ingestionTimeStart,
             DbQueryClient queryClient,
             DbCommandClient dbCommandClient,
             CancellationToken ct)
         {
-            var activityItem = RowItemGateway.InMemoryCache
-                .ActivityMap[iterationItem.ActivityName]
-                .RowItem;
-            var activityParam = Parameterization.Activities[iterationItem.ActivityName];
             var distributions = await queryClient.GetRecordDistributionAsync(
-                new KustoPriority(iterationItem.ActivityName, iterationItem.IterationId),
-                activityItem.SourceTable.TableName,
-                activityParam.KqlQuery,
-                iterationItem.CursorStart,
-                iterationItem.CursorEnd,
+                sourceTableItem.IterationId,
+                sourceTableItem.SourceTable.TableName,
+                sourceTableItem.CursorStart,
+                sourceTableItem.CursorEnd,
                 ingestionTimeStart,
                 MAX_STATS_COUNT,
                 ct);
@@ -175,8 +170,8 @@ private async Task<IImmutableList<RecordDistributionInExtent>> GetRecordDistribu
                     .Where(id => !string.IsNullOrWhiteSpace(id))
                     .Distinct();
                 var extentDates = await dbCommandClient.GetExtentDatesAsync(
-                    new KustoPriority(iterationItem.ActivityName, iterationItem.IterationId),
-                    activityItem.SourceTable.TableName,
+                    sourceTableItem.IterationId,
+                    sourceTableItem.SourceTable.TableName,
                     extentIds,
                     ct);
 
@@ -200,7 +195,7 @@ private async Task<IImmutableList<RecordDistributionInExtent>> GetRecordDistribu
                 else
                 {
                     return await GetRecordDistributionInExtents(
-                        iterationItem,
+                        sourceTableItem,
                         ingestionTimeStart,
                         queryClient,
                         dbCommandClient,