Parquet archive visibility and scheduled compaction

Lite/Database/DuckDbInitializer.cs

@@ -21,12 +21,24 @@
     /// </summary>
     internal const int CurrentSchemaVersion = 10;
 
+    private readonly string _archivePath;
+
     public DuckDbInitializer(string databasePath, ILogger<DuckDbInitializer>? logger = null)
     {
         _databasePath = databasePath;
         _logger = logger;
+        _archivePath = Path.Combine(Path.GetDirectoryName(databasePath) ?? ".", "archive");
     }
 
+    /* Tables that have parquet archives — views are created to UNION hot data with archived parquet files */
+    private static readonly string[] ArchivableTables =
+    [
+        "wait_stats", "query_stats", "procedure_stats", "query_store_stats",
+        "query_snapshots", "cpu_utilization_stats", "file_io_stats", "memory_stats",
+        "memory_clerks", "tempdb_stats", "perfmon_stats", "deadlocks",
+        "blocked_process_reports", "collection_log"
+    ];
+
     /// <summary>
     /// Gets the connection string for the DuckDB database.
     /// Disables automatic WAL checkpoints to prevent 2-3s stop-the-world stalls
@@ -40,20 +52,20 @@
     /// </summary>
     public async Task InitializeAsync()
     {
         _logger?.LogInformation("Initializing DuckDB database at {Path}", _databasePath);
 
         var directory = Path.GetDirectoryName(_databasePath);
         if (!string.IsNullOrEmpty(directory) && !Directory.Exists(directory))
         {
             Directory.CreateDirectory(directory);
             _logger?.LogInformation("Created database directory: {Directory}", directory);
         }
 
         var archivePath = Path.Combine(directory ?? ".", "archive");
         if (!Directory.Exists(archivePath))
         {
             Directory.CreateDirectory(archivePath);
             _logger?.LogInformation("Created archive directory: {ArchivePath}", archivePath);
         }
 
         /* Try to open the database. If the DuckDB storage version has changed,
@@ -82,7 +94,7 @@
 
             if (existingVersion < CurrentSchemaVersion)
             {
                 _logger?.LogInformation("Schema upgrade needed: v{Old} -> v{New}", existingVersion, CurrentSchemaVersion);
                 await RunMigrationsAsync(connection, existingVersion);
                 await SetSchemaVersionAsync(connection, CurrentSchemaVersion);
             }
@@ -97,8 +109,10 @@
                 await ExecuteNonQueryAsync(connection, indexStatement);
             }
 
             _logger?.LogInformation("Database initialization complete. Schema version: {Version}", CurrentSchemaVersion);
         }
+
+        await CreateArchiveViewsAsync();
     }
 
     /// <summary>
@@ -148,7 +162,7 @@
                 cmd.CommandText = $"EXPORT DATABASE '{exportDir.Replace("'", "''")}' (FORMAT PARQUET)";
                 await cmd.ExecuteNonQueryAsync();
                 exported = true;
                 _logger?.LogInformation("Exported old database to {ExportDir}", exportDir);
             }
         }
         catch (Exception ex)
@@ -162,7 +176,7 @@
         {
             /* DuckDB may have .wal files too */
             File.Move(_databasePath, backupPath);
             _logger?.LogInformation("Backed up old database to {BackupPath}", backupPath);
 
             var walPath = _databasePath + ".wal";
             if (File.Exists(walPath))
@@ -392,12 +406,12 @@
                 }
 
                 if (serverNames.Count > 0)
                     _logger?.LogInformation("Fixed server_id in {Table} for {Count} server(s)", table, serverNames.Count);
             }
             catch (Exception ex)
             {
                 /* Table might not exist yet — that's fine, it will be created with correct IDs */
                 _logger?.LogDebug(ex, "Skipped server_id fix for {Table} (may not exist yet)", table);
             }
         }
     }
@@ -410,6 +424,58 @@
         return new DuckDBConnection(ConnectionString);
     }
 
+    /// <summary>
+    /// Creates or refreshes views that UNION hot DuckDB tables with archived parquet files.
+    /// Call at startup and after each archive cycle so newly archived data is queryable.
+    /// </summary>
+    public async Task CreateArchiveViewsAsync()
+    {
+        using var connection = CreateConnection();
+        await connection.OpenAsync();
+
+        foreach (var table in ArchivableTables)
+        {
+            try
+            {
+                var parquetGlob = Path.Combine(_archivePath, $"*_{table}.parquet");
+                var hasParquetFiles = Directory.Exists(_archivePath)
+                    && Directory.GetFiles(_archivePath, $"*_{table}.parquet").Length > 0;
+
+                string viewSql;
+                if (hasParquetFiles)
+                {
+                    var globPath = parquetGlob.Replace("\\", "/");
+                    viewSql = $"CREATE OR REPLACE VIEW v_{table} AS SELECT * FROM {table} UNION ALL SELECT * FROM read_parquet('{globPath}', union_by_name=true)";
+                }
+                else
+                {
+                    viewSql = $"CREATE OR REPLACE VIEW v_{table} AS SELECT * FROM {table}";
+                }
+
+                using var cmd = connection.CreateCommand();
+                cmd.CommandText = viewSql;
+                await cmd.ExecuteNonQueryAsync();
+            }
+            catch (Exception ex)
+            {
+                /* Schema mismatch between hot table and old parquet — fall back to table-only view */
+                _logger?.LogWarning(ex, "Failed to create archive view for {Table}, using table-only view", table);
+                try
+                {
+                    using var fallbackCmd = connection.CreateCommand();
+                    fallbackCmd.CommandText = $"CREATE OR REPLACE VIEW v_{table} AS SELECT * FROM {table}";
+                    await fallbackCmd.ExecuteNonQueryAsync();
+                }
+                catch (Exception fallbackEx)
+                {
+                    _logger?.LogError(fallbackEx, "Failed to create fallback view for {Table}", table);
+                }
+            }
+        }
+
+        _logger?.LogDebug("Archive views created/refreshed for {Count} tables", ArchivableTables.Length);
+    }
+
     /// <summary>
     /// Runs a manual WAL checkpoint. Call this between collection cycles
     /// to flush the WAL during idle time instead of during collector writes.
@@ -470,4 +536,127 @@
         return fileInfo.Length / (1024.0 * 1024.0);
     }
 
+    /// <summary>
+    /// Compacts the database by exporting all tables to a fresh file and swapping.
+    /// DuckDB VACUUM does not reclaim space from append-fragmented files — only
+    /// export/reimport eliminates bloat. Typically takes 2-5 seconds for a 300MB database.
+    /// </summary>
+    /// <returns>True if compaction was performed, false if skipped or failed.</returns>
+    public async Task<bool> CompactAsync()
+    {
+        if (!DatabaseExists())
+        {
+            return false;
+        }
+
+        var sizeBefore = GetDatabaseSizeMb();
+        var tempPath = _databasePath + ".compact";
+        var backupPath = _databasePath + ".precompact";
+
+        _logger?.LogInformation("Starting database compaction ({SizeMb:F0} MB)", sizeBefore);
+
+        try
+        {
+            /* Export all data to a fresh database via ATTACH + CREATE TABLE AS */
+            if (File.Exists(tempPath)) File.Delete(tempPath);
+
+            using (var connection = CreateConnection())
+            {
+                await connection.OpenAsync();
+
+                /* Checkpoint first to flush WAL */
+                using (var cmd = connection.CreateCommand())
+                {
+                    cmd.CommandText = "CHECKPOINT";
+                    await cmd.ExecuteNonQueryAsync();
+                }
+
+                /* Attach the new database and copy all tables */
+                using (var cmd = connection.CreateCommand())
+                {
+                    cmd.CommandText = $"ATTACH '{tempPath.Replace("\\", "/")}' AS compact_db";
+                    await cmd.ExecuteNonQueryAsync();
+                }
+
+                /* Get all table names (exclude views) */
+                var tableNames = new List<string>();
+                using (var cmd = connection.CreateCommand())
+                {
+                    cmd.CommandText = "SELECT table_name FROM information_schema.tables WHERE table_schema = 'main' AND table_type = 'BASE TABLE'";
+                    using var reader = await cmd.ExecuteReaderAsync();
+                    while (await reader.ReadAsync())
+                    {
+                        tableNames.Add(reader.GetString(0));
+                    }
+                }
+
+                foreach (var table in tableNames)
+                {
+                    using var cmd = connection.CreateCommand();
+                    cmd.CommandText = $"CREATE TABLE compact_db.{table} AS SELECT * FROM main.{table}";
+                    await cmd.ExecuteNonQueryAsync();
+                }
+
+                using (var cmd = connection.CreateCommand())
+                {
+                    cmd.CommandText = "DETACH compact_db";
+                    await cmd.ExecuteNonQueryAsync();
+                }
+            }
+
+            /* Swap files: old → backup, compact → primary */
+            if (File.Exists(backupPath)) File.Delete(backupPath);
+            File.Move(_databasePath, backupPath);
+
+            var walPath = _databasePath + ".wal";
+            if (File.Exists(walPath)) File.Delete(walPath);
+
+            File.Move(tempPath, _databasePath);
+
+            /* Recreate indexes and views on the fresh database */
+            using (var connection = CreateConnection())
+            {
+                await connection.OpenAsync();
+
+                foreach (var indexStatement in Schema.GetAllIndexStatements())
+                {
+                    try
+                    {
+                        using var cmd = connection.CreateCommand();
+                        cmd.CommandText = indexStatement;
+                        await cmd.ExecuteNonQueryAsync();
+                    }
+                    catch { /* Index may already exist from CREATE TABLE AS */ }
+                }
+            }
+
+            await CreateArchiveViewsAsync();
+
+            /* Clean up backup */
+            File.Delete(backupPath);
+
+            var sizeAfter = GetDatabaseSizeMb();
+            _logger?.LogInformation("Compaction complete: {Before:F0} MB -> {After:F0} MB ({Saved:F0} MB reclaimed)",
+                sizeBefore, sizeAfter, sizeBefore - sizeAfter);
+
+            return true;
+        }
+        catch (Exception ex)
+        {
+            _logger?.LogError(ex, "Database compaction failed");
+
+            /* Restore from backup if the primary file was moved */
+            if (!File.Exists(_databasePath) && File.Exists(backupPath))
+            {
+                File.Move(backupPath, _databasePath);
+                _logger?.LogInformation("Restored database from pre-compaction backup");
+            }
+
+            /* Clean up temp file */
+            if (File.Exists(tempPath)) File.Delete(tempPath);
+
+            return false;
+        }
+    }
+
 }

Lite/MainWindow.xaml.cs

@@ -105,7 +105,7 @@ private async void MainWindow_Loaded(object sender, RoutedEventArgs e)
             var archiveService = new ArchiveService(_databaseInitializer, App.ArchiveDirectory);
             var retentionService = new RetentionService(App.ArchiveDirectory);
 
-            _backgroundService = new CollectionBackgroundService(_collectorService, archiveService, retentionService, _serverManager);
+            _backgroundService = new CollectionBackgroundService(_collectorService, _databaseInitializer, archiveService, retentionService, _serverManager);
 
             // Start background collection
             _backgroundCts = new CancellationTokenSource();

Lite/Services/ArchiveService.cs

@@ -128,6 +128,9 @@ UNION ALL
                 _logger?.LogError(ex, "Failed to archive table {Table}", table);
             }
         }
+
+        /* Refresh archive views so newly archived parquet files are queryable */
+        await _duckDb.CreateArchiveViewsAsync();
         }
         finally
         {

Lite/Services/CollectionBackgroundService.cs

@@ -22,6 +22,7 @@ namespace PerformanceMonitorLite.Services;
 public class CollectionBackgroundService : BackgroundService
 {
     private readonly RemoteCollectorService _collectorService;
+    private readonly DuckDbInitializer? _duckDb;
     private readonly ServerManager? _serverManager;
     private readonly ArchiveService? _archiveService;
     private readonly RetentionService? _retentionService;
@@ -30,23 +31,30 @@ public class CollectionBackgroundService : BackgroundService
     private static readonly TimeSpan CollectionInterval = TimeSpan.FromMinutes(1);
     private DateTime _lastArchiveTime = DateTime.MinValue;
     private DateTime _lastRetentionTime = DateTime.MinValue;
+    private DateTime _lastCompactionTime = DateTime.MinValue;
 
-    /* Archive every hour, retention cleanup once per day */
+    /* Archive every hour, retention + compaction once per day */
     private static readonly TimeSpan ArchiveInterval = TimeSpan.FromHours(1);
     private static readonly TimeSpan RetentionInterval = TimeSpan.FromHours(24);
+    private static readonly TimeSpan CompactionInterval = TimeSpan.FromHours(24);
+
+    /* Warn if database exceeds this size between compaction cycles */
+    private const double SizeWarningThresholdMb = 1024;
 
     public bool IsPaused { get; set; }
     public DateTime? LastCollectionTime { get; private set; }
     public bool IsCollecting { get; private set; }
 
     public CollectionBackgroundService(
         RemoteCollectorService collectorService,
+        DuckDbInitializer? duckDb = null,
         ArchiveService? archiveService = null,
         RetentionService? retentionService = null,
         ServerManager? serverManager = null,
         ILogger<CollectionBackgroundService>? logger = null)
     {
         _collectorService = collectorService;
+        _duckDb = duckDb;
         _serverManager = serverManager;
         _archiveService = archiveService;
         _retentionService = retentionService;
@@ -108,6 +116,9 @@ stall collectors mid-write with 2-3s stop-the-world pauses */
 
                 /* Periodic retention cleanup */
                 RunRetentionIfDue();
+
+                /* Periodic database compaction to prevent bloat */
+                await RunCompactionIfDueAsync();
             }
 
             try
@@ -158,4 +169,37 @@ private void RunRetentionIfDue()
             _logger?.LogError(ex, "Retention cleanup failed");
         }
     }
+
+    private async Task RunCompactionIfDueAsync()
+    {
+        if (_duckDb == null || DateTime.UtcNow - _lastCompactionTime < CompactionInterval)
+        {
+            /* Size watchdog: warn if database is large even between compaction cycles */
+            if (_duckDb != null)
+            {
+                var sizeMb = _duckDb.GetDatabaseSizeMb();
+                if (sizeMb > SizeWarningThresholdMb)
+                {
+                    _logger?.LogWarning("Database size is {SizeMb:F0} MB (threshold: {Threshold} MB) — compaction will run at next scheduled interval",
+                        sizeMb, SizeWarningThresholdMb);
+                }
+            }
+            return;
+        }
+
+        try
+        {
+            IsPaused = true;
+            await _duckDb.CompactAsync();
+            _lastCompactionTime = DateTime.UtcNow;
+        }
+        catch (Exception ex)
+        {
+            _logger?.LogError(ex, "Database compaction failed");
+        }
+        finally
+        {
+            IsPaused = false;
+        }
+    }
 }

Lite/Services/LocalDataService.Blocking.cs

@@ -78,7 +78,7 @@ public async Task<List<DeadlockRow>> GetRecentDeadlocksAsync(int serverId, int h
     victim_process_id,
     victim_sql_text,
     deadlock_graph_xml
-FROM deadlocks
+FROM v_deadlocks
 WHERE server_id = $1
 AND   collection_time >= $2
 AND   collection_time <= $3
@@ -139,7 +139,7 @@ public async Task<List<QuerySnapshotRow>> GetLatestQuerySnapshotsAsync(int serve
     query_plan,
     live_query_plan,
     collection_time
-FROM query_snapshots
+FROM v_query_snapshots
 WHERE server_id = $1
 AND   collection_time >= $2
 AND   collection_time <= $3
@@ -230,7 +230,7 @@ public async Task<List<BlockedProcessReportRow>> GetRecentBlockedProcessReportsA
     blocking_last_batch_completed,
     blocked_priority,
     blocking_priority
-FROM blocked_process_reports
+FROM v_blocked_process_reports
 WHERE server_id = $1
 AND   collection_time >= $2
 AND   collection_time <= $3
@@ -310,7 +310,7 @@ Group by event_time (when blocking actually occurred) rather than collection_tim
     SELECT
         DATE_TRUNC('minute', event_time) AS bucket,
         COUNT(*) AS incident_count
-    FROM blocked_process_reports
+    FROM v_blocked_process_reports
     WHERE server_id = $1
     AND   event_time >= $2
     AND   event_time <= $3
@@ -353,7 +353,7 @@ public async Task<List<TrendPoint>> GetDeadlockTrendAsync(int serverId, int hour
     SELECT
         DATE_TRUNC('hour', deadlock_time) AS bucket,
         COUNT(*) AS deadlock_count
-    FROM deadlocks
+    FROM v_deadlocks
     WHERE server_id = $1
     AND   collection_time >= $2
     AND   collection_time <= $3

Lite/Services/LocalDataService.CollectionHealth.cs

@@ -34,7 +34,7 @@ public async Task<List<CollectorHealthRow>> GetCollectionHealthAsync(int serverI
     MAX(CASE WHEN status IN ('ERROR', 'PERMISSIONS') THEN error_message END) AS last_error,
     MAX(CASE WHEN status IN ('ERROR', 'PERMISSIONS') THEN collection_time END) AS last_error_time,
     SUM(CASE WHEN status = 'PERMISSIONS' THEN 1 ELSE 0 END) AS permission_denied_count
-FROM collection_log
+FROM v_collection_log
 WHERE server_id = $1
 AND   collection_time >= $2
 GROUP BY collector_name
@@ -83,7 +83,7 @@ public async Task<List<CollectionLogRow>> GetRecentCollectionLogAsync(int server
     status,
     error_message,
     server_name
-FROM collection_log
+FROM v_collection_log
 WHERE server_id = $1
 AND   collection_time >= $2
 ORDER BY collection_time DESC

Lite/Services/LocalDataService.Cpu.cs

@@ -32,7 +32,7 @@ public async Task<List<CpuUtilizationRow>> GetCpuUtilizationAsync(int serverId,
     sample_time,
     sqlserver_cpu_utilization,
     other_process_cpu_utilization
-FROM cpu_utilization_stats
+FROM v_cpu_utilization_stats
 WHERE server_id = $1
 AND   sample_time >= $2
 AND   sample_time <= $3