.

Author: vplauzon

code/KustoCopyConsole/CommandLineOptions.cs

@@ -25,7 +25,7 @@ public class CommandLineOptions
         public string Destination { get; set; } = string.Empty;
 
         [Option('a', "auth", Required = false, HelpText = "Set authentication method.")]
-        public string Authentication { get; set; } = "AzCli";
+        public string Authentication { get; set; } = string.Empty;
 
         [Option(
             'l',

code/KustoCopyConsole/Entity/InMemory/RowItemInMemoryCache.cs

@@ -10,63 +10,77 @@ namespace KustoCopyConsole.Entity.InMemory
     internal class RowItemInMemoryCache
     {
         private readonly object _lock = new object();
-        private volatile IImmutableDictionary<(Uri, string), SourceDatabaseCache> _sourceDatabaseMap
-            = ImmutableDictionary<(Uri, string), SourceDatabaseCache>.Empty;
+        private volatile IImmutableDictionary<TableIdentity, SourceTableCache> _sourceTableMap =
+            ImmutableDictionary<TableIdentity, SourceTableCache>.Empty;
 
         public RowItemInMemoryCache(IEnumerable<RowItem> items)
         {
             lock (_lock)
             {
                 foreach (var item in items)
                 {
-                    AddItem(item);
+                    AppendItem(item);
                 }
             }
         }
 
-        public IImmutableDictionary<(Uri, string), SourceDatabaseCache> SoureDatabaseMap
-            => _sourceDatabaseMap;
+        public IImmutableDictionary<TableIdentity, SourceTableCache> SourceTableMap
+            => _sourceTableMap;
 
         public IEnumerable<RowItem> GetItems()
         {
-            throw new NotImplementedException();
+            foreach (var sourceTable in SourceTableMap.Values)
+            {
+                foreach (var sourceTableIteration in sourceTable.IterationMap.Values)
+                {
+                    yield return sourceTableIteration.RowItem;
+                }
+            }
         }
 
-        public void AddItem(RowItem item)
+        public void AppendItem(RowItem item)
         {
             lock (_lock)
             {
                 if (item.RowType != RowType.FileVersion && item.RowType != RowType.Unspecified)
                 {
                     Interlocked.Exchange(
-                        ref _sourceDatabaseMap,
-                        AddItemToCache(item));
+                        ref _sourceTableMap,
+                        AppendItemToCache(item));
                 }
             }
         }
 
-        private IImmutableDictionary<(Uri, string), SourceDatabaseCache> AddItemToCache(RowItem item)
+        private IImmutableDictionary<TableIdentity, SourceTableCache> AppendItemToCache(
+            RowItem item)
         {
             switch (item.RowType)
             {
-                case RowType.SourceDatabase:
-                    return AddSourceDatabase(item);
+                case RowType.SourceTable:
+                    return AppendSourceTable(item);
                 default:
                     throw new NotSupportedException($"Not supported row type:  {item.RowType}");
             }
         }
 
-        private IImmutableDictionary<(Uri, string), SourceDatabaseCache> AddSourceDatabase(RowItem item)
+        private IImmutableDictionary<TableIdentity, SourceTableCache> AppendSourceTable(
+            RowItem item)
         {
-            var key = (NormalizedUri.NormalizeUri(item.SourceClusterUri), item.SourceDatabaseName);
+                //item.IterationId!.Value
+            var tableId = new TableIdentity(
+                NormalizedUri.NormalizeUri(item.SourceClusterUri),
+                item.SourceDatabaseName,
+                item.SourceTableName);
 
-            if (_sourceDatabaseMap.ContainsKey(key))
+            if (_sourceTableMap.ContainsKey(tableId))
             {
-                return _sourceDatabaseMap.SetItem(key, new SourceDatabaseCache(item));
+                return _sourceTableMap.SetItem(
+                    tableId,
+                    _sourceTableMap[tableId].AppendIteration(item));
             }
             else
             {
-                return _sourceDatabaseMap.Add(key, new SourceDatabaseCache(item));
+                return _sourceTableMap.Add(tableId, new SourceTableCache(item));
             }
         }
     }

code/KustoCopyConsole/Entity/InMemory/SourceDatabaseCache.cs

@@ -0,0 +1,40 @@
+using System.Collections.Immutable;
+
+namespace KustoCopyConsole.Entity.InMemory
+{
+    internal class SourceTableCache
+    {
+        public SourceTableCache(
+            IImmutableDictionary<long, SourceTableIterationCache> iterationMap)
+        {
+            IterationMap = iterationMap;
+        }
+
+        public SourceTableCache(RowItem iterationItem)
+        {
+            var iterationId = iterationItem.IterationId;
+
+            IterationMap = ImmutableDictionary<long, SourceTableIterationCache>
+                .Empty
+                .Add(iterationId, new SourceTableIterationCache(iterationItem));
+        }
+
+        public IImmutableDictionary<long, SourceTableIterationCache> IterationMap { get; }
+
+        public SourceTableCache AppendIteration(RowItem item)
+        {
+            var iterationId = item.IterationId;
+
+            if (IterationMap.ContainsKey(iterationId))
+            {
+                return new SourceTableCache(
+                    IterationMap.SetItem(iterationId, IterationMap[iterationId].Update(item)));
+            }
+            else
+            {
+                return new SourceTableCache(
+                    IterationMap.Add(iterationId, new SourceTableIterationCache(item)));
+            }
+        }
+    }
+}

code/KustoCopyConsole/Entity/InMemory/SourceTableCache.cs

@@ -0,0 +1,17 @@
+using System.Collections.Immutable;
+
+namespace KustoCopyConsole.Entity.InMemory
+{
+    internal class SourceTableIterationCache : CacheBase
+    {
+        public SourceTableIterationCache(RowItem item)
+            : base(item)
+        {
+        }
+
+        public SourceTableIterationCache Update(RowItem item)
+        {
+            return new SourceTableIterationCache(item);
+        }
+    }
+}

code/KustoCopyConsole/Entity/InMemory/SourceTableIterationCache.cs

@@ -0,0 +1,13 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace KustoCopyConsole.Entity.InMemory
+{
+    internal record TableIdentity(
+        Uri ClusterUri,
+        string DatabaseName,
+        string TableName);
+}

code/KustoCopyConsole/Entity/InMemory/TableIdentity.cs

@@ -85,32 +85,35 @@ private class RowTypeConverter : DefaultTypeConverter
         public string SourceDatabaseName { get; set; } = string.Empty;
 
         [Index(7)]
-        public string DestinationClusterUri { get; set; } = string.Empty;
+        public string SourceTableName { get; set; } = string.Empty;
 
         [Index(8)]
-        public string DestinationDatabaseName { get; set; } = string.Empty;
+        public string DestinationClusterUri { get; set; } = string.Empty;
 
         [Index(9)]
-        public string TableName { get; set; } = string.Empty;
+        public string DestinationDatabaseName { get; set; } = string.Empty;
 
-        /// <summary>Zero-based index with zero being the backfill.</summary>>
         [Index(10)]
-        public long? IterationId { get; set; }
+        public string DestinationTableName { get; set; } = string.Empty;
 
+        /// <summary>Zero-based index with zero being the backfill.</summary>>
         [Index(11)]
-        public string CursorStart { get; set; } = string.Empty;
+        public long IterationId { get; set; }
 
         [Index(12)]
-        public string CursorEnd { get; set; } = string.Empty;
+        public string CursorStart { get; set; } = string.Empty;
 
         [Index(13)]
-        public long? BlockId { get; set; }
+        public string CursorEnd { get; set; } = string.Empty;
 
         [Index(14)]
-        public DateTime? IngestionTimeStart { get; set; }
+        public long? BlockId { get; set; }
 
         [Index(15)]
-        public long? IngestionTimeEnd { get; set; }
+        public DateTime? IngestionTimeStart { get; set; }
+
+        [Index(16)]
+        public DateTime? IngestionTimeEnd { get; set; }
         #endregion
 
         public T ParseState<T>() where T : struct

code/KustoCopyConsole/Entity/RowItem.cs

@@ -4,6 +4,6 @@ public enum RowType
     {
         Unspecified,
         FileVersion,
-        SourceDatabase
+        SourceTable
     }
 }

code/KustoCopyConsole/Entity/RowType.cs

@@ -2,7 +2,6 @@
 {
     public enum SourceTableState
     {
-        Discovered,
         Planning,
         Exporting,
         Completed

code/KustoCopyConsole/Entity/State/DestinationDatabaseState.cs

@@ -0,0 +1,33 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace KustoCopyConsole
+{
+    internal static class EnumerableHelper
+    {
+        public static T ArgMax<T>(this IEnumerable<T> enumerable, Func<T, long> selector)
+        {
+            T? maxItem = default;
+            long maxValue = 0;
+
+            foreach (var item in enumerable)
+            {
+                if (maxItem == null
+                    || selector(item) > maxValue)
+                {
+                    maxItem = item;
+                }
+            }
+
+            if (maxItem == null)
+            {
+                throw new InvalidOperationException("Enumerable is empty");
+            }
+
+            return maxItem;
+        }
+    }
+}

code/KustoCopyConsole/Entity/State/SourceDatabaseState.cs

@@ -44,9 +44,9 @@ private MainOrchestration(
 
         private static TokenCredential CreateCredentials(string authentication)
         {
-            if (string.Compare(authentication.Trim(), "AzCli", true) == 0)
+            if (string.IsNullOrWhiteSpace(authentication))
             {
-                return new AzureCliCredential();
+                return new DefaultAzureCredential();
             }
             else
             {