Ingestion progress tracking and reporting + tests

Wildcard index bug fix

Author: kristan66

src/KeeperData.Core/ETL/Impl/IngestionPipeline.cs

@@ -2,6 +2,7 @@
 using CsvHelper.Configuration;
 using KeeperData.Core.Database;
 using KeeperData.Core.ETL.Abstract;
+using KeeperData.Core.ETL.Utils;
 using KeeperData.Core.Reporting;
 using KeeperData.Core.Reporting.Dtos;
 using KeeperData.Core.Storage;
@@ -25,12 +26,14 @@ public class IngestionPipeline(
     IMongoClient mongoClient,
     IOptions<IDatabaseConfig> databaseConfig,
     IImportReportingService reportingService,
+    CsvRowCounter csvRowCounter,
     ILogger<IngestionPipeline> logger) : IIngestionPipeline
 {
     private const int BatchSize = 1000;
     private const int LogInterval = 100;
     private const int LineageEventBatchSize = 500;
     private readonly IDatabaseConfig _databaseConfig = databaseConfig.Value;
+    private readonly CsvRowCounter _rowCounter = csvRowCounter;
 
     // MongoDB field name constants
     private const string FieldId = "_id";
@@ -112,7 +115,8 @@ private async Task UpdateIngestionPhaseStartedAsync(Guid importId, CancellationT
             FilesProcessed = 0,
             RecordsCreated = 0,
             RecordsUpdated = 0,
-            RecordsDeleted = 0
+            RecordsDeleted = 0,
+            CurrentFileStatus = null
         }, ct);
     }
 
@@ -151,11 +155,12 @@ CancellationToken ct
 
                 totals = totals.Add(fileResult);
 
-                // Update progress after each file
+                // Clear current file status after completion and update overall progress
                 await UpdateIngestionPhaseProgressAsync(
                     importId,
                     processedFileCount,
                     totals,
+                    null, // Clear current file status
                     ct);
             }
         }
@@ -236,6 +241,8 @@ private async Task<FileIngestionMetrics> IngestFileAsync(
         await EnsureWildcardIndexExistsAsync(collection, ct);
 
         string? tempFilePath = null;
+        IngestionProgressTracker? progressTracker = null;
+
         try
         {
             // Track S3 download time
@@ -246,6 +253,13 @@ private async Task<FileIngestionMetrics> IngestFileAsync(
             logger.LogInformation("Downloaded file {FileKey} to temp storage: {TempPath} in {DownloadDuration}ms",
                 file.Key, tempFilePath, downloadStopwatch.ElapsedMilliseconds);
 
+            // Count rows for progress tracking
+            var estimatedRowCount = await _rowCounter.CountRowsAsync(tempFilePath, ct);
+            progressTracker = new IngestionProgressTracker(file.Key, estimatedRowCount);
+
+            logger.LogInformation("File {FileKey} has approximately {RowCount} data rows to process",
+                file.Key, estimatedRowCount);
+
             // Track MongoDB ingestion time
             var mongoIngestionStopwatch = Stopwatch.StartNew();
 
@@ -265,6 +279,7 @@ private async Task<FileIngestionMetrics> IngestFileAsync(
                 file.Key,
                 collectionName,
                 fileSet.Definition,
+                progressTracker,
                 ct);
 
             await csvContext.DisposeAsync();
@@ -418,12 +433,14 @@ private async Task<FileIngestionMetrics> ProcessCsvRecordsAsync(
         string fileKey,
         string collectionName,
         DataSetDefinition definition,
+        IngestionProgressTracker progressTracker,
         CancellationToken ct)
     {
         var metrics = new RecordMetricsAccumulator();
         var batch = new List<(BsonDocument Document, string ChangeType)>();
         var lineageEvents = new List<RecordLineageEvent>();
         var totalMongoProcessingMs = 0L;
+        var totals = new IngestionTotals();
 
         while (await csv.ReadAsync())
         {
@@ -459,8 +476,27 @@ private async Task<FileIngestionMetrics> ProcessCsvRecordsAsync(
                 totalMongoProcessingMs += batchStopwatch.ElapsedMilliseconds;
 
                 metrics.AddBatch(batchMetrics);
+                totals = totals.Add(new IngestionTotals
+                {
+                    RecordsCreated = batchMetrics.RecordsCreated,
+                    RecordsUpdated = batchMetrics.RecordsUpdated,
+                    RecordsDeleted = batchMetrics.RecordsDeleted
+                });
+
+                // Update progress tracking and report every 100 records
+                progressTracker.UpdateProgress(metrics.RecordsProcessed);
 
-                LogProgressIfNeeded(metrics.RecordsProcessed, fileKey);
+                if (metrics.RecordsProcessed % LogInterval == 0)
+                {
+                    LogProgressIfNeeded(metrics.RecordsProcessed, fileKey);
+
+                    var currentStatus = progressTracker.GetCurrentStatus();
+                    await UpdateIngestionPhaseProgressWithFileStatusAsync(
+                        importId,
+                        totals,
+                        currentStatus,
+                        ct);
+                }
 
                 batch.Clear();
 
@@ -482,6 +518,16 @@ private async Task<FileIngestionMetrics> ProcessCsvRecordsAsync(
                 ct);
 
             metrics.AddBatch(batchMetrics);
+
+            // Final progress update
+            progressTracker.UpdateProgress(metrics.RecordsProcessed);
+            var finalStatus = progressTracker.Complete();
+
+            await UpdateIngestionPhaseProgressWithFileStatusAsync(
+                importId,
+                totals,
+                finalStatus,
+                ct);
         }
 
         // Flush remaining lineage events
@@ -597,7 +643,7 @@ private async Task CreateWildcardIndexAsync(
         logger.LogInformation("Creating wildcard index on collection {CollectionName}",
             collection.CollectionNamespace.CollectionName);
 
-        var wildcardIndexKeys = Builders<BsonDocument>.IndexKeys.Wildcard("$**");
+        var wildcardIndexKeys = Builders<BsonDocument>.IndexKeys.Wildcard();
         var indexModel = new CreateIndexModel<BsonDocument>(wildcardIndexKeys);
 
         await collection.Indexes.CreateOneAsync(indexModel, cancellationToken: ct);
@@ -944,6 +990,7 @@ private async Task UpdateIngestionPhaseProgressAsync(
         Guid importId,
         int filesProcessed,
         IngestionTotals totals,
+        IngestionCurrentFileStatus? currentFileStatus,
         CancellationToken ct)
     {
         await reportingService.UpdateIngestionPhaseAsync(importId, new IngestionPhaseUpdate
@@ -952,7 +999,25 @@ private async Task UpdateIngestionPhaseProgressAsync(
             FilesProcessed = filesProcessed,
             RecordsCreated = totals.RecordsCreated,
             RecordsUpdated = totals.RecordsUpdated,
-            RecordsDeleted = totals.RecordsDeleted
+            RecordsDeleted = totals.RecordsDeleted,
+            CurrentFileStatus = currentFileStatus
+        }, ct);
+    }
+
+    private async Task UpdateIngestionPhaseProgressWithFileStatusAsync(
+        Guid importId,
+        IngestionTotals totals,
+        IngestionCurrentFileStatus currentFileStatus,
+        CancellationToken ct)
+    {
+        await reportingService.UpdateIngestionPhaseAsync(importId, new IngestionPhaseUpdate
+        {
+            Status = PhaseStatus.Started,
+            FilesProcessed = 0, // Not updated during file processing
+            RecordsCreated = totals.RecordsCreated,
+            RecordsUpdated = totals.RecordsUpdated,
+            RecordsDeleted = totals.RecordsDeleted,
+            CurrentFileStatus = currentFileStatus
         }, ct);
     }
 

src/KeeperData.Core/ETL/Setup/ServiceCollectionExtensions.cs

@@ -1,5 +1,6 @@
 using KeeperData.Core.ETL.Abstract;
 using KeeperData.Core.ETL.Impl;
+using KeeperData.Core.ETL.Utils;
 using KeeperData.Core.Reporting.Setup;
 using Microsoft.Extensions.DependencyInjection;
 using System.Diagnostics.CodeAnalysis;
@@ -12,6 +13,7 @@ public static class ServiceCollectionExtensions
     public static void AddEtlDependencies(this IServiceCollection services)
     {
         services.AddSingleton<IDataSetDefinitions>(_ => StandardDataSetDefinitionsBuilder.Build());
+        services.AddTransient<CsvRowCounter>();
         services.AddTransient<IExternalCatalogueServiceFactory, ExternalCatalogueServiceFactory>();
         services.AddTransient<IIngestionPipeline, IngestionPipeline>();
         services.AddTransient<IAcquisitionPipeline, AcquisitionPipeline>();

src/KeeperData.Core/ETL/Utils/CsvRowCounter.cs

@@ -0,0 +1,122 @@
+using Microsoft.Extensions.Logging;
+
+namespace KeeperData.Core.ETL.Utils;
+
+/// <summary>
+/// Utility class for counting rows in CSV files efficiently.
+/// </summary>
+public class CsvRowCounter
+{
+    private readonly ILogger<CsvRowCounter> _logger;
+    private const int BufferSize = 65536; // 64KB buffer for efficient reading
+    private const int MaxRetryAttempts = 3;
+    private const int RetryDelayMs = 100;
+
+    public CsvRowCounter(ILogger<CsvRowCounter> logger)
+    {
+        _logger = logger;
+    }
+
+    /// <summary>
+    /// Counts the approximate number of data rows in a CSV file (excluding header).
+    /// This is a fast approximation that counts newlines and may not be 100% accurate
+    /// for CSVs with quoted fields containing newlines.
+    /// </summary>
+    public async Task<int> CountRowsAsync(string filePath, CancellationToken ct)
+    {
+        for (int attempt = 1; attempt <= MaxRetryAttempts; attempt++)
+        {
+            try
+            {
+                return await CountRowsInternalAsync(filePath, ct);
+            }
+            catch (Exception ex) when (attempt < MaxRetryAttempts)
+            {
+                _logger.LogWarning(ex, "Failed to count rows in file {FilePath} on attempt {Attempt}/{MaxAttempts}. Retrying...",
+                    filePath, attempt, MaxRetryAttempts);
+
+                await Task.Delay(RetryDelayMs * attempt, ct);
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex, "Failed to count rows in file {FilePath} after {MaxAttempts} attempts",
+                    filePath, MaxRetryAttempts);
+                throw;
+            }
+        }
+
+        // This should never be reached due to the throw in the catch block
+        return 0;
+    }
+
+    private async Task<int> CountRowsInternalAsync(string filePath, CancellationToken ct)
+    {
+        if (!File.Exists(filePath))
+        {
+            throw new FileNotFoundException($"CSV file not found: {filePath}");
+        }
+
+        var fileInfo = new FileInfo(filePath);
+
+        // Handle empty files
+        if (fileInfo.Length == 0)
+        {
+            _logger.LogDebug("File {FilePath} is empty, returning 0 rows", filePath);
+            return 0;
+        }
+
+        await using var fileStream = new FileStream(
+            filePath,
+            FileMode.Open,
+            FileAccess.Read,
+            FileShare.Read,
+            BufferSize,
+            useAsync: true);
+
+        var buffer = new byte[BufferSize];
+        var rowCount = 0;
+        var bytesRead = 0;
+        var lastCharWasNewline = false;
+
+        while ((bytesRead = await fileStream.ReadAsync(buffer, 0, buffer.Length, ct)) > 0)
+        {
+            for (int i = 0; i < bytesRead; i++)
+            {
+                if (buffer[i] == '\n')
+                {
+                    rowCount++;
+                    lastCharWasNewline = true;
+                }
+                else if (buffer[i] == '\r')
+                {
+                    // Handle Windows line endings (\r\n)
+                    // Don't double-count if next char is \n
+                    if (i + 1 < bytesRead && buffer[i + 1] == '\n')
+                    {
+                        i++; // Skip the \n
+                    }
+                    rowCount++;
+                    lastCharWasNewline = true;
+                }
+                else
+                {
+                    lastCharWasNewline = false;
+                }
+            }
+        }
+
+        // If file doesn't end with newline, we still have a row
+        if (!lastCharWasNewline && fileInfo.Length > 0)
+        {
+            rowCount++;
+        }
+
+        // Subtract 1 for header row (if file has at least one row)
+        var dataRows = Math.Max(0, rowCount - 1);
+
+        _logger.LogDebug("Counted approximately {RowCount} data rows in file {FilePath} (excluding header)",
+            dataRows, filePath);
+
+        return dataRows;
+    }
+}