Skip to content

Harden DuckDB queries and fix IsArchiving race condition #850

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
erikdarlingdata merged 1 commit into from
Apr 15, 2026
Merged

Harden DuckDB queries and fix IsArchiving race condition #850

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 6 additions & 2 deletions Lite/Database/DuckDbInitializer.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -135,13 +135,13 @@
/// </summary>
public async Task InitializeAsync()
{
_logger?.LogInformation("Initializing DuckDB database at {Path}", _databasePath);

Check warning on line 138 in Lite/Database/DuckDbInitializer.cs

View workflow job for this annotation

GitHub Actions / build 

Evaluation of this argument may be expensive and unnecessary if logging is disabled (https://learn.microsoft.com/dotnet/fundamentals/code-analysis/quality-rules/ca1873)

var directory = Path.GetDirectoryName(_databasePath);
if (!string.IsNullOrEmpty(directory) && !Directory.Exists(directory))
{
Directory.CreateDirectory(directory);
_logger?.LogInformation("Created database directory: {Directory}", directory);

Check warning on line 144 in Lite/Database/DuckDbInitializer.cs

View workflow job for this annotation

GitHub Actions / build 

Evaluation of this argument may be expensive and unnecessary if logging is disabled (https://learn.microsoft.com/dotnet/fundamentals/code-analysis/quality-rules/ca1873)
}

var archivePath = Path.Combine(directory ?? ".", "archive");
Expand Down Expand Up @@ -254,7 +254,7 @@
cmd.CommandText = $"EXPORT DATABASE '{exportDir.Replace("'", "''")}' (FORMAT PARQUET)";
await cmd.ExecuteNonQueryAsync();
exported = true;
_logger?.LogInformation("Exported old database to {ExportDir}", exportDir);

Check warning on line 257 in Lite/Database/DuckDbInitializer.cs

View workflow job for this annotation

GitHub Actions / build 

Evaluation of this argument may be expensive and unnecessary if logging is disabled (https://learn.microsoft.com/dotnet/fundamentals/code-analysis/quality-rules/ca1873)
}
}
catch (Exception ex)
Expand All @@ -268,7 +268,7 @@
{
/* DuckDB may have .wal files too */
File.Move(_databasePath, backupPath);
_logger?.LogInformation("Backed up old database to {BackupPath}", backupPath);

Check warning on line 271 in Lite/Database/DuckDbInitializer.cs

View workflow job for this annotation

GitHub Actions / build 

Evaluation of this argument may be expensive and unnecessary if logging is disabled (https://learn.microsoft.com/dotnet/fundamentals/code-analysis/quality-rules/ca1873)

var walPath = _databasePath + ".wal";
if (File.Exists(walPath))
Expand Down Expand Up @@ -724,7 +724,7 @@
string viewSql;
if (hasParquetFiles)
{
var globPath = parquetGlob.Replace("\\", "/");
var globPath = EscapeSqlPath(parquetGlob.Replace("\\", "/"));
if (table == "config_alert_log")
{
viewSql = $@"CREATE OR REPLACE VIEW v_{table} AS
Expand Down Expand Up @@ -935,5 +935,9 @@
await InitializeAsync();
}


/// <summary>
/// Escapes single quotes in a file path for safe interpolation into DuckDB SQL.
/// DuckDB does not support parameterized paths in read_parquet() or COPY TO.
/// </summary>
internal static string EscapeSqlPath(string path) => path.Replace("'", "''");
}
33 changes: 22 additions & 11 deletions Lite/Services/ArchiveService.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,8 +33,14 @@ public class ArchiveService
/// <summary>
/// Indicates whether an archival operation is currently in progress.
/// UI code can check this to warn users before dismiss or show a status indicator.
/// Volatile-backed to ensure cross-thread visibility without locking.
/// </summary>
public static bool IsArchiving { get; private set; }
private static volatile bool s_isArchiving;
public static bool IsArchiving
{
get => s_isArchiving;
private set => s_isArchiving = value;
}

/* Tables eligible for archival with their time column.
IMPORTANT: Every table with time-series data must be listed here,
Expand Down Expand Up @@ -133,7 +139,8 @@ Archive views use glob (*_table.parquet) to pick up all files. */

/* Delete archived rows from hot table */
using var deleteCmd = connection.CreateCommand();
deleteCmd.CommandText = $"DELETE FROM {table} WHERE {timeColumn} < '{cutoffDate:yyyy-MM-dd HH:mm:ss}'";
deleteCmd.CommandText = $"DELETE FROM {table} WHERE {timeColumn} < $1";
deleteCmd.Parameters.Add(new DuckDBParameter { Value = cutoffDate });
await deleteCmd.ExecuteNonQueryAsync();

_logger?.LogInformation("Archived {Count} rows from {Table} to {Path}", rowCount, table, parquetPath);
Expand Down Expand Up @@ -161,7 +168,8 @@ Archive views use glob (*_table.parquet) to pick up all files. */
private static async Task<long> GetRowCountBeforeCutoff(DuckDBConnection connection, string table, string timeColumn, DateTime cutoff)
{
using var cmd = connection.CreateCommand();
cmd.CommandText = $"SELECT COUNT(*) FROM {table} WHERE {timeColumn} < '{cutoff:yyyy-MM-dd HH:mm:ss}'";
cmd.CommandText = $"SELECT COUNT(*) FROM {table} WHERE {timeColumn} < $1";
cmd.Parameters.Add(new DuckDBParameter { Value = cutoff });
var result = await cmd.ExecuteScalarAsync();
return Convert.ToInt64(result);
}
Expand All @@ -171,11 +179,14 @@ private static async Task ExportToParquet(DuckDBConnection connection, string ta
using var cmd = connection.CreateCommand();
cmd.CommandText = $@"
COPY (
SELECT * FROM {table} WHERE {timeColumn} < '{cutoff:yyyy-MM-dd HH:mm:ss}'
) TO '{filePath}' (FORMAT PARQUET, COMPRESSION ZSTD)";
SELECT * FROM {table} WHERE {timeColumn} < $1
) TO '{EscapeSqlPath(filePath)}' (FORMAT PARQUET, COMPRESSION ZSTD)";
cmd.Parameters.Add(new DuckDBParameter { Value = cutoff });
await cmd.ExecuteNonQueryAsync();
}

private static string EscapeSqlPath(string path) => DuckDbInitializer.EscapeSqlPath(path);

/* Columns to exclude during compaction — dead weight from legacy archives */
private static readonly Dictionary<string, string[]> CompactionExcludeColumns = new()
{
Expand Down Expand Up @@ -343,7 +354,7 @@ Each group gets its own DuckDB connection so memory is fully released between gr
{
using var schemaCon = new DuckDBConnection("DataSource=:memory:");
schemaCon.Open();
var allPathList = string.Join(", ", sourcePaths.Select(p => $"'{p}'"));
var allPathList = string.Join(", ", sourcePaths.Select(p => $"'{EscapeSqlPath(p)}'"));
using var schemaCmd = schemaCon.CreateCommand();
schemaCmd.CommandText = $"SELECT column_name FROM (DESCRIBE SELECT * FROM read_parquet([{allPathList}], union_by_name=true))";
using var reader = schemaCmd.ExecuteReader();
Expand All @@ -368,10 +379,10 @@ Each group gets its own DuckDB connection so memory is fully released between gr
pragma.ExecuteNonQuery();
}

var pathList = string.Join(", ", sourcePaths.Select(p => $"'{p}'"));
var pathList = string.Join(", ", sourcePaths.Select(p => $"'{EscapeSqlPath(p)}'"));
using var cmd = con.CreateCommand();
cmd.CommandText = $"COPY (SELECT {selectClause} FROM read_parquet([{pathList}], union_by_name=true)) " +
$"TO '{tempPath}' (FORMAT PARQUET, COMPRESSION ZSTD, ROW_GROUP_SIZE 122880)";
$"TO '{EscapeSqlPath(tempPath)}' (FORMAT PARQUET, COMPRESSION ZSTD, ROW_GROUP_SIZE 122880)";
cmd.ExecuteNonQuery();
}
else
Expand Down Expand Up @@ -399,10 +410,10 @@ Sort smallest-first so early merges are cheap. */
pragma.ExecuteNonQuery();
}

var pairList = $"'{currentPath}', '{sorted[i]}'";
var pairList = $"'{EscapeSqlPath(currentPath)}', '{EscapeSqlPath(sorted[i])}'";
using var cmd = con.CreateCommand();
cmd.CommandText = $"COPY (SELECT {selectClause} FROM read_parquet([{pairList}], union_by_name=true)) " +
$"TO '{stepOutput}' (FORMAT PARQUET, COMPRESSION ZSTD, ROW_GROUP_SIZE 122880)";
$"TO '{EscapeSqlPath(stepOutput)}' (FORMAT PARQUET, COMPRESSION ZSTD, ROW_GROUP_SIZE 122880)";
cmd.ExecuteNonQuery();

/* Clean up previous intermediate file */
Expand Down Expand Up @@ -511,7 +522,7 @@ Archive views use glob (*_table.parquet) to pick up all files. */
.Replace("\\", "/");

using var exportCmd = connection.CreateCommand();
exportCmd.CommandText = $"COPY (SELECT * FROM {table}) TO '{parquetPath}' (FORMAT PARQUET, COMPRESSION ZSTD)";
exportCmd.CommandText = $"COPY (SELECT * FROM {table}) TO '{EscapeSqlPath(parquetPath)}' (FORMAT PARQUET, COMPRESSION ZSTD)";
await exportCmd.ExecuteNonQueryAsync();

_logger?.LogInformation("Archived {Count} rows from {Table}", rowCount, table);
Expand Down
Loading