diff --git a/Cortex.sln b/Cortex.sln index ec50673..cca087c 100644 --- a/Cortex.sln +++ b/Cortex.sln @@ -72,6 +72,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Serialization", "Serializat EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Cortex.Streams.Mediator", "src\Cortex.Streams.Mediator\Cortex.Streams.Mediator.csproj", "{84410C57-0F59-F31F-B921-4C1F3D3FF144}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Cortex.States.DuckDb", "src\Cortex.States.DuckDb\Cortex.States.DuckDb.csproj", "{4FAE6C5E-53EE-4CCE-85A6-B7551A92C488}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -202,6 +204,10 @@ Global {84410C57-0F59-F31F-B921-4C1F3D3FF144}.Debug|Any CPU.Build.0 = Debug|Any CPU {84410C57-0F59-F31F-B921-4C1F3D3FF144}.Release|Any CPU.ActiveCfg = Release|Any CPU {84410C57-0F59-F31F-B921-4C1F3D3FF144}.Release|Any CPU.Build.0 = Release|Any CPU + {4FAE6C5E-53EE-4CCE-85A6-B7551A92C488}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {4FAE6C5E-53EE-4CCE-85A6-B7551A92C488}.Debug|Any CPU.Build.0 = Debug|Any CPU + {4FAE6C5E-53EE-4CCE-85A6-B7551A92C488}.Release|Any CPU.ActiveCfg = Release|Any CPU + {4FAE6C5E-53EE-4CCE-85A6-B7551A92C488}.Release|Any CPU.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE @@ -233,6 +239,7 @@ Global {44A166BD-01E9-4A4B-9BC5-7DE01B472E73} = {1C5D462D-168D-4D3F-B96E-CCE5517DB197} {472BC645-9E2F-4205-A571-4D9184747EC5} = {7F9E0AEA-721E-46F8-90ED-8EA8423647FB} {84410C57-0F59-F31F-B921-4C1F3D3FF144} = {4C68702C-1661-4AD9-83FD-E0B52B791969} + {4FAE6C5E-53EE-4CCE-85A6-B7551A92C488} = {C31F8C0F-8BCF-4959-9BA1-8645D058EAA0} EndGlobalSection GlobalSection(ExtensibilityGlobals) = postSolution SolutionGuid = {E20303B6-8AC9-4FFF-B645-4608309ADA94} diff --git a/README.md b/README.md index f6fa9ee..307a537 100644 --- a/README.md +++ b/README.md @@ -114,6 +114,9 @@ - **Cortex.States.SQLite:** Persistent state storage using SQLite. [![NuGet Version](https://img.shields.io/nuget/v/Cortex.States.SQLite?label=Cortex.States.SQLite)](https://www.nuget.org/packages/Cortex.States.SQLite) +- **Cortex.States.DuckDb:** Persistent state storage using DuckDb. +[![NuGet Version](https://img.shields.io/nuget/v/Cortex.States.DuckDb?label=Cortex.States.DuckDb)](https://www.nuget.org/packages/Cortex.States.DuckDb) + - **Cortex.Telemetry:** Core library to add support for Tracing and Matrics. [![NuGet Version](https://img.shields.io/nuget/v/Cortex.Telemetry?label=Cortex.Telemetry)](https://www.nuget.org/packages/Cortex.Telemetry) diff --git a/docs/Cortex.States.DuckDb.md b/docs/Cortex.States.DuckDb.md new file mode 100644 index 0000000..5c58877 --- /dev/null +++ b/docs/Cortex.States.DuckDb.md @@ -0,0 +1,310 @@ +# Cortex.States.DuckDb + +[![NuGet Version](https://img.shields.io/nuget/v/Cortex.States.DuckDb?label=Cortex.States.DuckDb)](https://www.nuget.org/packages/Cortex.States.DuckDb) + +**Cortex.States.DuckDb** is a state store implementation for the Cortex Data Framework that uses [DuckDB](https://duckdb.org/) as the underlying storage engine. DuckDB is an in-process analytical database management system designed for fast analytical queries, making it an excellent choice for scenarios requiring both transactional state management and analytical capabilities. + +## Features + +- **High-Performance Analytics**: Leverages DuckDB's columnar storage and vectorized query execution +- **In-Memory & Persistent Storage**: Supports both in-memory databases for fast processing and file-based persistence +- **Native Export Capabilities**: Export data directly to Parquet or CSV formats +- **Batch Operations**: Efficient bulk insert and delete operations with transaction support +- **Thread-Safe**: Built-in thread safety for concurrent access +- **Flexible Serialization**: Customizable key and value serialization +- **Fluent Builder API**: Easy configuration through builder pattern + +## Installation + +### Using the .NET CLI + +```bash +dotnet add package Cortex.States.DuckDb +``` + +### Using the Package Manager Console + +```powershell +Install-Package Cortex.States.DuckDb +``` + +## Quick Start + +### Basic Usage + +```csharp +using Cortex.States.DuckDb; + +// Create a persistent DuckDB state store +var stateStore = new DuckDbKeyValueStateStore( + name: "MyStateStore", + databasePath: "./data/mystore.duckdb", + tableName: "KeyValueStore" +); + +// Store values +stateStore.Put("counter", 42); +stateStore.Put("total", 100); + +// Retrieve values +var counter = stateStore.Get("counter"); // Returns 42 + +// Check if key exists +if (stateStore.ContainsKey("counter")) +{ + Console.WriteLine("Counter exists!"); +} + +// Remove a value +stateStore.Remove("counter"); + +// Get all keys +foreach (var key in stateStore.GetKeys()) +{ + Console.WriteLine($"Key: {key}"); +} + +// Don't forget to dispose +stateStore.Dispose(); +``` + +### Using the Fluent Builder + +```csharp +using Cortex.States.DuckDb; + +// Create store using fluent builder +var stateStore = DuckDbKeyValueStateStoreBuilder + .Create("OrderStore") + .WithDatabasePath("./data/orders.duckdb") + .WithTableName("Orders") + .WithIndex(true) + .WithMaxMemory("2GB") + .WithThreads(4) + .Build(); + +// Use the store +stateStore.Put("ORD-001", new OrderSummary { Total = 99.99m, Status = "Completed" }); +``` + +### In-Memory Database + +```csharp +using Cortex.States.DuckDb; + +// Create an in-memory store for fast processing +var inMemoryStore = DuckDbKeyValueStateStoreBuilder + .Create("TemporaryStore") + .UseInMemory() + .WithTableName("TempData") + .Build(); + +// Perfect for temporary computations +inMemoryStore.Put("sum", 1234.56m); +``` + +### Using with Options + +```csharp +using Cortex.States.DuckDb; + +// Create options for fine-grained control +var options = new DuckDbKeyValueStateStoreOptions +{ + DatabasePath = "./data/analytics.duckdb", + TableName = "AnalyticsState", + CreateIndex = true, + MaxMemory = "4GB", + Threads = 8, + AccessMode = DuckDbAccessMode.ReadWrite +}; + +var stateStore = new DuckDbKeyValueStateStore( + name: "AnalyticsStore", + options: options +); +``` + +### Factory Methods + +```csharp +using Cortex.States.DuckDb; + +// Quick creation methods +var persistentStore = DuckDbStateStoreExtensions + .CreatePersistentDuckDbStore("ProductStore", "./data/products.duckdb", "Products"); + +var inMemoryStore = DuckDbStateStoreExtensions + .CreateInMemoryDuckDbStore("SessionStore", "Sessions"); +``` + +## Advanced Features + +### Batch Operations + +```csharp +// Efficient bulk insert +var items = new List> +{ + new("price-1", 10.99m), + new("price-2", 20.99m), + new("price-3", 30.99m) +}; + +stateStore.PutMany(items); + +// Bulk delete +stateStore.RemoveMany(new[] { "price-1", "price-2" }); +``` + +### Export to Parquet/CSV + +DuckDB has native support for Parquet and CSV formats, making data export seamless: + +```csharp +// Export to Parquet (ideal for analytics) +stateStore.ExportToParquet("./exports/state-backup.parquet"); + +// Export to CSV (ideal for data sharing) +stateStore.ExportToCsv("./exports/state-backup.csv"); +``` + +### Count and Clear + +```csharp +// Get total count +var count = stateStore.Count(); +Console.WriteLine($"Total items: {count}"); + +// Clear all items +stateStore.Clear(); +``` + +### Checkpoint + +For persistent databases, you can force a checkpoint to ensure all data is written to disk: + +```csharp +stateStore.Checkpoint(); +``` + +## Integration with Cortex Streams + +Use DuckDB state store with Cortex Streams for stateful stream processing: + +```csharp +using Cortex.Streams; +using Cortex.States.DuckDb; + +// Create the state store +var stateStore = new DuckDbKeyValueStateStore( + name: "WordCountStore", + databasePath: "./data/wordcount.duckdb", + tableName: "WordCounts" +); + +// Use in a stream pipeline +var stream = StreamBuilder.CreateNewStream("WordCountStream") + .Stream() + .FlatMap(line => line.Split(' ')) + .GroupBy(word => word) + .Aggregate( + stateStore, + (count, word) => count + 1, + initialValue: 0) + .Sink(result => Console.WriteLine($"{result.Key}: {result.Value}")) + .Build(); + +stream.Start(); +``` + +## Custom Serialization + +You can provide custom serializers for complex types: + +```csharp +using System.Text.Json; + +var stateStore = new DuckDbKeyValueStateStore( + name: "ComplexStore", + databasePath: "./data/complex.duckdb", + tableName: "ComplexData", + keySerializer: key => key.ToString(), + keyDeserializer: str => Guid.Parse(str), + valueSerializer: value => JsonSerializer.Serialize(value), + valueDeserializer: str => JsonSerializer.Deserialize(str)! +); +``` + +## Configuration Options + +| Option | Description | Default | +|--------|-------------|---------| +| `DatabasePath` | Path to the DuckDB database file. Use `:memory:` for in-memory | Required | +| `TableName` | Name of the table for key-value storage | Required | +| `UseInMemory` | Use in-memory database instead of file | `false` | +| `CreateIndex` | Create index on key column for faster lookups | `true` | +| `MaxMemory` | Maximum memory limit (e.g., "1GB", "512MB") | Auto | +| `Threads` | Number of threads (0 = auto) | `0` | +| `AccessMode` | Database access mode (Automatic, ReadWrite, ReadOnly) | `Automatic` | + +## When to Use DuckDB State Store + +DuckDB is particularly well-suited for: + +- **Analytical workloads**: When you need to run analytical queries on your state +- **Large datasets**: Efficient columnar storage for large amounts of data +- **Data export requirements**: Native Parquet/CSV export capabilities +- **Embedded analytics**: In-process database without external dependencies +- **Temporary processing**: Fast in-memory mode for intermediate computations + +Consider other state stores when: + +- You need distributed state across multiple nodes (use Cassandra, MongoDB) +- You require extreme write throughput (use RocksDB) +- You need full ACID transactions across multiple operations (use PostgreSQL, SQL Server) + +## Thread Safety + +The `DuckDbKeyValueStateStore` is thread-safe and can be used concurrently from multiple threads. For in-memory databases, a persistent connection is maintained to ensure data consistency. + +## Error Handling + +```csharp +try +{ + var value = stateStore.Get("non-existent-key"); + if (value == null) + { + Console.WriteLine("Key not found"); + } +} +catch (InvalidOperationException ex) +{ + Console.WriteLine($"Store not initialized: {ex.Message}"); +} +``` + +## Best Practices + +1. **Dispose properly**: Always dispose of the state store when done to release resources +2. **Use batch operations**: For bulk inserts/deletes, use `PutMany` and `RemoveMany` +3. **Choose appropriate storage**: Use in-memory for temporary data, file-based for persistence +4. **Set memory limits**: Configure `MaxMemory` for large datasets to prevent excessive memory usage +5. **Regular checkpoints**: Call `Checkpoint()` periodically for critical data in persistent mode + +## Requirements + +- .NET 7.0 or later +- DuckDB.NET.Data package (automatically included) + +## License + +MIT License - see the [license file](../src/Cortex.States.DuckDb/Assets/license.md) for details. + +## Related Packages + +- [Cortex.States](https://www.nuget.org/packages/Cortex.States) - Core state management +- [Cortex.States.RocksDb](https://www.nuget.org/packages/Cortex.States.RocksDb) - RocksDB state store +- [Cortex.States.SQLite](https://www.nuget.org/packages/Cortex.States.SQLite) - SQLite state store +- [Cortex.Streams](https://www.nuget.org/packages/Cortex.Streams) - Core streaming capabilities diff --git a/src/Cortex.States.DuckDb/Assets/cortex.png b/src/Cortex.States.DuckDb/Assets/cortex.png new file mode 100644 index 0000000..101a1fb Binary files /dev/null and b/src/Cortex.States.DuckDb/Assets/cortex.png differ diff --git a/src/Cortex.States.DuckDb/Assets/license.md b/src/Cortex.States.DuckDb/Assets/license.md new file mode 100644 index 0000000..caa98b4 --- /dev/null +++ b/src/Cortex.States.DuckDb/Assets/license.md @@ -0,0 +1,20 @@ +The MIT License (MIT) + +Copyright (c) 2026 Buildersoft + +Permission is hereby granted, free of charge, to any person obtaining a copy of +this software and associated documentation files (the "Software"), to deal in +the Software without restriction, including without limitation the rights to +use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of +the Software, and to permit persons to whom the Software is furnished to do so, +subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS +FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR +COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER +IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN +CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. diff --git a/src/Cortex.States.DuckDb/Cortex.States.DuckDb.csproj b/src/Cortex.States.DuckDb/Cortex.States.DuckDb.csproj new file mode 100644 index 0000000..dfca926 --- /dev/null +++ b/src/Cortex.States.DuckDb/Cortex.States.DuckDb.csproj @@ -0,0 +1,54 @@ + + + + net9.0;net8.0;net7.0 + + 3.0.0 + 3.0.0 + Buildersoft Cortex Framework + Buildersoft + Buildersoft,EnesHoxha + Copyright © Buildersoft 2026 + + Cortex Data Framework is a robust, extensible platform designed to facilitate real-time data streaming, processing, and state management. It provides developers with a comprehensive suite of tools and libraries to build scalable, high-performance data pipelines tailored to diverse use cases. By abstracting underlying streaming technologies and state management solutions, Cortex Data Framework enables seamless integration, simplified development workflows, and enhanced maintainability for complex data-driven applications. + + https://github.com/buildersoftio/cortex + cortex mediator eda streaming distributed streams states duckdb analytics olap + + 3.0.0 + license.md + cortex.png + Cortex.States.DuckDb + True + True + True + + Just as the Cortex in our brains handles complex processing efficiently, Cortex Data Framework brings brainpower to your data management! + https://buildersoft.io/ + README.md + + + + + True + \ + + + True + + + + True + + + + + + + + + + + + + diff --git a/src/Cortex.States.DuckDb/DuckDbKeyValueStateStore.cs b/src/Cortex.States.DuckDb/DuckDbKeyValueStateStore.cs new file mode 100644 index 0000000..116801a --- /dev/null +++ b/src/Cortex.States.DuckDb/DuckDbKeyValueStateStore.cs @@ -0,0 +1,670 @@ +using DuckDB.NET.Data; +using System; +using System.Collections.Generic; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; + +namespace Cortex.States.DuckDb +{ + /// + /// A key-value state store implementation backed by DuckDB. + /// DuckDB is an in-process analytical database management system designed for fast analytics. + /// + /// The type of keys in the store. + /// The type of values in the store. + public class DuckDbKeyValueStateStore : IDataStore, IDisposable + { + private readonly string _connectionString; + private readonly string _tableName; + private readonly Func _keySerializer; + private readonly Func _valueSerializer; + private readonly Func _keyDeserializer; + private readonly Func _valueDeserializer; + private readonly DuckDbKeyValueStateStoreOptions _options; + private readonly DuckDBConnection _persistentConnection; + private readonly object _connectionLock = new object(); + + private static readonly SemaphoreSlim _initializationLock = new SemaphoreSlim(1, 1); + private volatile bool _isInitialized; + private bool _disposed; + + /// + /// Gets the name of the state store. + /// + public string Name { get; } + + /// + /// Initializes a new instance of the DuckDbKeyValueStateStore. + /// + /// A friendly name for the store. + /// + /// The file path to the DuckDB database. + /// Use ":memory:" for an in-memory database, or provide a file path for persistence. + /// + /// The name of the table to use for storing state entries. + /// Optional key serializer. If not provided, JSON serialization is used. + /// Optional value serializer. If not provided, JSON serialization is used. + /// Optional key deserializer. If not provided, JSON deserialization is used. + /// Optional value deserializer. If not provided, JSON deserialization is used. + public DuckDbKeyValueStateStore( + string name, + string databasePath, + string tableName, + Func keySerializer = null, + Func valueSerializer = null, + Func keyDeserializer = null, + Func valueDeserializer = null) + : this(name, new DuckDbKeyValueStateStoreOptions + { + DatabasePath = databasePath, + TableName = tableName + }, keySerializer, valueSerializer, keyDeserializer, valueDeserializer) + { + } + + /// + /// Initializes a new instance of the DuckDbKeyValueStateStore with options. + /// + /// A friendly name for the store. + /// Configuration options for the DuckDB state store. + /// Optional key serializer. If not provided, JSON serialization is used. + /// Optional value serializer. If not provided, JSON serialization is used. + /// Optional key deserializer. If not provided, JSON deserialization is used. + /// Optional value deserializer. If not provided, JSON deserialization is used. + public DuckDbKeyValueStateStore( + string name, + DuckDbKeyValueStateStoreOptions options, + Func keySerializer = null, + Func valueSerializer = null, + Func keyDeserializer = null, + Func valueDeserializer = null) + { + if (string.IsNullOrWhiteSpace(name)) + throw new ArgumentNullException(nameof(name)); + if (options == null) + throw new ArgumentNullException(nameof(options)); + if (string.IsNullOrWhiteSpace(options.DatabasePath)) + throw new ArgumentException("DatabasePath is required", nameof(options)); + if (string.IsNullOrWhiteSpace(options.TableName)) + throw new ArgumentException("TableName is required", nameof(options)); + + Name = name; + _options = options; + _tableName = options.TableName; + + // Build connection string + _connectionString = BuildConnectionString(options); + + // Assign custom or default (JSON-based) serializers/deserializers + _keySerializer = keySerializer ?? (key => JsonSerializer.Serialize(key)); + _valueSerializer = valueSerializer ?? (value => JsonSerializer.Serialize(value)); + _keyDeserializer = keyDeserializer ?? (str => JsonSerializer.Deserialize(str)); + _valueDeserializer = valueDeserializer ?? (str => JsonSerializer.Deserialize(str)); + + // Create a persistent connection for in-memory databases + if (options.UseInMemory || options.DatabasePath == ":memory:") + { + _persistentConnection = new DuckDBConnection(_connectionString); + _persistentConnection.Open(); + } + + // Initialize the table + InitializeAsync().GetAwaiter().GetResult(); + } + + private static string BuildConnectionString(DuckDbKeyValueStateStoreOptions options) + { + if (options.UseInMemory || options.DatabasePath == ":memory:") + { + return "DataSource=:memory:"; + } + + return $"DataSource={options.DatabasePath}"; + } + + private DuckDBConnection GetConnection() + { + if (_persistentConnection != null) + { + return _persistentConnection; + } + + var connection = new DuckDBConnection(_connectionString); + connection.Open(); + return connection; + } + + private void ReleaseConnection(DuckDBConnection connection) + { + // Only close and dispose if it's not the persistent connection + if (connection != _persistentConnection) + { + connection.Close(); + connection.Dispose(); + } + } + + private async Task InitializeAsync() + { + if (_isInitialized) return; + + await _initializationLock.WaitAsync().ConfigureAwait(false); + try + { + if (_isInitialized) return; + + var connection = GetConnection(); + try + { + // Create the table if it does not exist + var createTableSql = $@" + CREATE TABLE IF NOT EXISTS ""{_tableName}"" ( + key VARCHAR PRIMARY KEY, + value VARCHAR + );"; + + using (var cmd = connection.CreateCommand()) + { + cmd.CommandText = createTableSql; + await cmd.ExecuteNonQueryAsync().ConfigureAwait(false); + } + + // Create index for faster lookups if configured + if (_options.CreateIndex) + { + var createIndexSql = $@" + CREATE INDEX IF NOT EXISTS idx_{_tableName}_key + ON ""{_tableName}"" (key);"; + + using (var cmd = connection.CreateCommand()) + { + cmd.CommandText = createIndexSql; + await cmd.ExecuteNonQueryAsync().ConfigureAwait(false); + } + } + } + finally + { + ReleaseConnection(connection); + } + + _isInitialized = true; + } + finally + { + _initializationLock.Release(); + } + } + + private void EnsureInitialized() + { + if (!_isInitialized) + { + throw new InvalidOperationException("DuckDbKeyValueStateStore is not properly initialized."); + } + } + + /// + /// Gets the value associated with the specified key. + /// + /// The key of the value to get. + /// The value associated with the specified key, or default if the key is not found. + public TValue Get(TKey key) + { + EnsureInitialized(); + + var serializedKey = _keySerializer(key); + var connection = GetConnection(); + + try + { + var sql = $@"SELECT value FROM ""{_tableName}"" WHERE key = $key;"; + using (var cmd = connection.CreateCommand()) + { + cmd.CommandText = sql; + cmd.Parameters.Add(new DuckDBParameter("key", serializedKey)); + + var result = cmd.ExecuteScalar(); + if (result == null || result == DBNull.Value) + return default; + + return _valueDeserializer(result.ToString()); + } + } + finally + { + ReleaseConnection(connection); + } + } + + /// + /// Adds or updates the value associated with the specified key. + /// + /// The key of the value to add or update. + /// The value to add or update. + public void Put(TKey key, TValue value) + { + EnsureInitialized(); + + var serializedKey = _keySerializer(key); + var serializedValue = _valueSerializer(value); + var connection = GetConnection(); + + try + { + // DuckDB supports INSERT OR REPLACE syntax + var sql = $@" + INSERT OR REPLACE INTO ""{_tableName}"" (key, value) + VALUES ($key, $value);"; + + using (var cmd = connection.CreateCommand()) + { + cmd.CommandText = sql; + cmd.Parameters.Add(new DuckDBParameter("key", serializedKey)); + cmd.Parameters.Add(new DuckDBParameter("value", serializedValue)); + cmd.ExecuteNonQuery(); + } + } + finally + { + ReleaseConnection(connection); + } + } + + /// + /// Determines whether the store contains the specified key. + /// + /// The key to locate in the store. + /// true if the store contains an element with the specified key; otherwise, false. + public bool ContainsKey(TKey key) + { + EnsureInitialized(); + + var serializedKey = _keySerializer(key); + var connection = GetConnection(); + + try + { + var sql = $@"SELECT COUNT(*) FROM ""{_tableName}"" WHERE key = $key;"; + using (var cmd = connection.CreateCommand()) + { + cmd.CommandText = sql; + cmd.Parameters.Add(new DuckDBParameter("key", serializedKey)); + + var count = Convert.ToInt64(cmd.ExecuteScalar()); + return count > 0; + } + } + finally + { + ReleaseConnection(connection); + } + } + + /// + /// Removes the value with the specified key from the store. + /// + /// The key of the element to remove. + public void Remove(TKey key) + { + EnsureInitialized(); + + var serializedKey = _keySerializer(key); + var connection = GetConnection(); + + try + { + var sql = $@"DELETE FROM ""{_tableName}"" WHERE key = $key;"; + using (var cmd = connection.CreateCommand()) + { + cmd.CommandText = sql; + cmd.Parameters.Add(new DuckDBParameter("key", serializedKey)); + cmd.ExecuteNonQuery(); + } + } + finally + { + ReleaseConnection(connection); + } + } + + /// + /// Returns all key-value pairs in the store. + /// + /// An enumerable of all key-value pairs in the store. + public IEnumerable> GetAll() + { + EnsureInitialized(); + + var results = new List>(); + var connection = GetConnection(); + + try + { + var sql = $@"SELECT key, value FROM ""{_tableName}"";"; + using (var cmd = connection.CreateCommand()) + { + cmd.CommandText = sql; + using (var reader = cmd.ExecuteReader()) + { + while (reader.Read()) + { + var serializedKey = reader.GetString(0); + var serializedValue = reader.IsDBNull(1) ? null : reader.GetString(1); + + var key = _keyDeserializer(serializedKey); + var value = serializedValue == null ? default : _valueDeserializer(serializedValue); + + results.Add(new KeyValuePair(key, value)); + } + } + } + + return results; + } + finally + { + ReleaseConnection(connection); + } + } + + /// + /// Returns all keys in the store. + /// + /// An enumerable of all keys in the store. + public IEnumerable GetKeys() + { + EnsureInitialized(); + + var results = new List(); + var connection = GetConnection(); + + try + { + var sql = $@"SELECT key FROM ""{_tableName}"";"; + using (var cmd = connection.CreateCommand()) + { + cmd.CommandText = sql; + using (var reader = cmd.ExecuteReader()) + { + while (reader.Read()) + { + var serializedKey = reader.GetString(0); + results.Add(_keyDeserializer(serializedKey)); + } + } + } + + return results; + } + finally + { + ReleaseConnection(connection); + } + } + + /// + /// Adds or updates multiple key-value pairs in a batch operation. + /// + /// The key-value pairs to add or update. + public void PutMany(IEnumerable> items) + { + EnsureInitialized(); + + var connection = GetConnection(); + + try + { + using (var transaction = connection.BeginTransaction()) + { + try + { + foreach (var item in items) + { + var serializedKey = _keySerializer(item.Key); + var serializedValue = _valueSerializer(item.Value); + + var sql = $@" + INSERT OR REPLACE INTO ""{_tableName}"" (key, value) + VALUES ($key, $value);"; + + using (var cmd = connection.CreateCommand()) + { + cmd.CommandText = sql; + cmd.Transaction = transaction; + cmd.Parameters.Add(new DuckDBParameter("key", serializedKey)); + cmd.Parameters.Add(new DuckDBParameter("value", serializedValue)); + cmd.ExecuteNonQuery(); + } + } + + transaction.Commit(); + } + catch + { + transaction.Rollback(); + throw; + } + } + } + finally + { + ReleaseConnection(connection); + } + } + + /// + /// Removes multiple keys from the store in a batch operation. + /// + /// The keys to remove. + public void RemoveMany(IEnumerable keys) + { + EnsureInitialized(); + + var connection = GetConnection(); + + try + { + using (var transaction = connection.BeginTransaction()) + { + try + { + foreach (var key in keys) + { + var serializedKey = _keySerializer(key); + + var sql = $@"DELETE FROM ""{_tableName}"" WHERE key = $key;"; + using (var cmd = connection.CreateCommand()) + { + cmd.CommandText = sql; + cmd.Transaction = transaction; + cmd.Parameters.Add(new DuckDBParameter("key", serializedKey)); + cmd.ExecuteNonQuery(); + } + } + + transaction.Commit(); + } + catch + { + transaction.Rollback(); + throw; + } + } + } + finally + { + ReleaseConnection(connection); + } + } + + /// + /// Gets the count of items in the store. + /// + /// The number of items in the store. + public long Count() + { + EnsureInitialized(); + + var connection = GetConnection(); + + try + { + var sql = $@"SELECT COUNT(*) FROM ""{_tableName}"";"; + using (var cmd = connection.CreateCommand()) + { + cmd.CommandText = sql; + return Convert.ToInt64(cmd.ExecuteScalar()); + } + } + finally + { + ReleaseConnection(connection); + } + } + + /// + /// Clears all items from the store. + /// + public void Clear() + { + EnsureInitialized(); + + var connection = GetConnection(); + + try + { + var sql = $@"DELETE FROM ""{_tableName}"";"; + using (var cmd = connection.CreateCommand()) + { + cmd.CommandText = sql; + cmd.ExecuteNonQuery(); + } + } + finally + { + ReleaseConnection(connection); + } + } + + /// + /// Exports the state store data to a Parquet file. + /// DuckDB has native support for Parquet format. + /// + /// The path to the Parquet file to create. + public void ExportToParquet(string filePath) + { + EnsureInitialized(); + + var connection = GetConnection(); + + try + { + var sql = $@"COPY ""{_tableName}"" TO '{filePath}' (FORMAT PARQUET);"; + using (var cmd = connection.CreateCommand()) + { + cmd.CommandText = sql; + cmd.ExecuteNonQuery(); + } + } + finally + { + ReleaseConnection(connection); + } + } + + /// + /// Exports the state store data to a CSV file. + /// + /// The path to the CSV file to create. + public void ExportToCsv(string filePath) + { + EnsureInitialized(); + + var connection = GetConnection(); + + try + { + var sql = $@"COPY ""{_tableName}"" TO '{filePath}' (FORMAT CSV, HEADER);"; + using (var cmd = connection.CreateCommand()) + { + cmd.CommandText = sql; + cmd.ExecuteNonQuery(); + } + } + finally + { + ReleaseConnection(connection); + } + } + + /// + /// Creates a checkpoint to ensure all data is written to disk. + /// Only applicable for persistent databases. + /// + public void Checkpoint() + { + if (_options.UseInMemory || _options.DatabasePath == ":memory:") + { + return; // No checkpoint needed for in-memory databases + } + + var connection = GetConnection(); + + try + { + using (var cmd = connection.CreateCommand()) + { + cmd.CommandText = "CHECKPOINT;"; + cmd.ExecuteNonQuery(); + } + } + finally + { + ReleaseConnection(connection); + } + } + + /// + /// Releases all resources used by the DuckDbKeyValueStateStore. + /// + public void Dispose() + { + Dispose(true); + GC.SuppressFinalize(this); + } + + /// + /// Releases the unmanaged resources used by the DuckDbKeyValueStateStore and optionally releases the managed resources. + /// + /// true to release both managed and unmanaged resources; false to release only unmanaged resources. + protected virtual void Dispose(bool disposing) + { + if (_disposed) + return; + + if (disposing) + { + try + { + // Create checkpoint before closing for persistent databases + if (!_options.UseInMemory && _options.DatabasePath != ":memory:") + { + Checkpoint(); + } + } + catch + { + // Ignore checkpoint errors during disposal + } + + _persistentConnection?.Close(); + _persistentConnection?.Dispose(); + _initializationLock?.Dispose(); + } + + _disposed = true; + } + } +} diff --git a/src/Cortex.States.DuckDb/DuckDbKeyValueStateStoreOptions.cs b/src/Cortex.States.DuckDb/DuckDbKeyValueStateStoreOptions.cs new file mode 100644 index 0000000..b6bd9b0 --- /dev/null +++ b/src/Cortex.States.DuckDb/DuckDbKeyValueStateStoreOptions.cs @@ -0,0 +1,137 @@ +using System; + +namespace Cortex.States.DuckDb +{ + /// + /// Configuration options for the DuckDB key-value state store. + /// + public class DuckDbKeyValueStateStoreOptions + { + /// + /// Gets or sets the path to the DuckDB database file. + /// Use ":memory:" for an in-memory database. + /// + /// + /// "./data/mystore.duckdb" for a file-based database + /// ":memory:" for an in-memory database + /// + public string DatabasePath { get; set; } + + /// + /// Gets or sets the name of the table to use for storing key-value pairs. + /// + public string TableName { get; set; } + + /// + /// Gets or sets a value indicating whether to use an in-memory database. + /// When true, the DatabasePath is ignored and ":memory:" is used. + /// Default is false. + /// + public bool UseInMemory { get; set; } = false; + + /// + /// Gets or sets a value indicating whether to create an index on the key column. + /// Improves lookup performance for large datasets. + /// Default is true. + /// + public bool CreateIndex { get; set; } = true; + + /// + /// Gets or sets the number of threads DuckDB should use. + /// Set to 0 to use all available threads. + /// Default is 0 (auto). + /// + public int Threads { get; set; } = 0; + + /// + /// Gets or sets the maximum memory limit for DuckDB. + /// Examples: "1GB", "512MB", "2GB" + /// Leave null for default (80% of system memory). + /// + public string MaxMemory { get; set; } + + /// + /// Gets or sets a value indicating whether to enable object cache. + /// Improves performance for repeated queries. + /// Default is true. + /// + public bool EnableObjectCache { get; set; } = true; + + /// + /// Gets or sets the access mode for the database. + /// + public DuckDbAccessMode AccessMode { get; set; } = DuckDbAccessMode.Automatic; + + /// + /// Creates a new instance of DuckDbKeyValueStateStoreOptions for an in-memory database. + /// + /// The name of the table to use for storing key-value pairs. + /// A new options instance configured for in-memory use. + public static DuckDbKeyValueStateStoreOptions InMemory(string tableName) + { + if (string.IsNullOrWhiteSpace(tableName)) + throw new ArgumentNullException(nameof(tableName)); + + return new DuckDbKeyValueStateStoreOptions + { + DatabasePath = ":memory:", + TableName = tableName, + UseInMemory = true + }; + } + + /// + /// Creates a new instance of DuckDbKeyValueStateStoreOptions for a file-based database. + /// + /// The path to the DuckDB database file. + /// The name of the table to use for storing key-value pairs. + /// A new options instance configured for file-based persistence. + public static DuckDbKeyValueStateStoreOptions Persistent(string databasePath, string tableName) + { + if (string.IsNullOrWhiteSpace(databasePath)) + throw new ArgumentNullException(nameof(databasePath)); + if (string.IsNullOrWhiteSpace(tableName)) + throw new ArgumentNullException(nameof(tableName)); + + return new DuckDbKeyValueStateStoreOptions + { + DatabasePath = databasePath, + TableName = tableName, + UseInMemory = false + }; + } + + /// + /// Validates the options and throws if they are invalid. + /// + public void Validate() + { + if (string.IsNullOrWhiteSpace(TableName)) + throw new ArgumentException("TableName is required", nameof(TableName)); + + if (!UseInMemory && string.IsNullOrWhiteSpace(DatabasePath)) + throw new ArgumentException("DatabasePath is required when not using in-memory mode", nameof(DatabasePath)); + } + } + + /// + /// Specifies the access mode for the DuckDB database. + /// + public enum DuckDbAccessMode + { + /// + /// DuckDB automatically determines the access mode. + /// + Automatic = 0, + + /// + /// Opens the database in read-write mode. + /// + ReadWrite = 1, + + /// + /// Opens the database in read-only mode. + /// + ReadOnly = 2 + } +} diff --git a/src/Cortex.States.DuckDb/DuckDbStateStoreExtensions.cs b/src/Cortex.States.DuckDb/DuckDbStateStoreExtensions.cs new file mode 100644 index 0000000..ec066df --- /dev/null +++ b/src/Cortex.States.DuckDb/DuckDbStateStoreExtensions.cs @@ -0,0 +1,253 @@ +using System; + +namespace Cortex.States.DuckDb +{ + /// + /// Extension methods for creating and configuring DuckDB state stores. + /// + public static class DuckDbStateStoreExtensions + { + /// + /// Creates a new DuckDB key-value state store with a persistent database. + /// + /// The type of keys in the store. + /// The type of values in the store. + /// A friendly name for the store. + /// The path to the DuckDB database file. + /// The name of the table to use for storing key-value pairs. + /// A new DuckDB key-value state store instance. + public static DuckDbKeyValueStateStore CreateDuckDbStore( + string name, + string databasePath, + string tableName) + { + return new DuckDbKeyValueStateStore(name, databasePath, tableName); + } + + /// + /// Creates a new DuckDB key-value state store with configuration options. + /// + /// The type of keys in the store. + /// The type of values in the store. + /// A friendly name for the store. + /// An action to configure the options. + /// A new DuckDB key-value state store instance. + public static DuckDbKeyValueStateStore CreateDuckDbStore( + string name, + Action configureOptions) + { + var options = new DuckDbKeyValueStateStoreOptions(); + configureOptions(options); + options.Validate(); + + return new DuckDbKeyValueStateStore(name, options); + } + + /// + /// Creates a new in-memory DuckDB key-value state store. + /// + /// The type of keys in the store. + /// The type of values in the store. + /// A friendly name for the store. + /// The name of the table to use for storing key-value pairs. + /// A new in-memory DuckDB key-value state store instance. + public static DuckDbKeyValueStateStore CreateInMemoryDuckDbStore( + string name, + string tableName) + { + var options = DuckDbKeyValueStateStoreOptions.InMemory(tableName); + return new DuckDbKeyValueStateStore(name, options); + } + + /// + /// Creates a new persistent DuckDB key-value state store. + /// + /// The type of keys in the store. + /// The type of values in the store. + /// A friendly name for the store. + /// The path to the DuckDB database file. + /// The name of the table to use for storing key-value pairs. + /// A new persistent DuckDB key-value state store instance. + public static DuckDbKeyValueStateStore CreatePersistentDuckDbStore( + string name, + string databasePath, + string tableName) + { + var options = DuckDbKeyValueStateStoreOptions.Persistent(databasePath, tableName); + return new DuckDbKeyValueStateStore(name, options); + } + } + + /// + /// Builder class for creating DuckDB key-value state stores with fluent configuration. + /// + /// The type of keys in the store. + /// The type of values in the store. + public class DuckDbKeyValueStateStoreBuilder + { + private string _name; + private readonly DuckDbKeyValueStateStoreOptions _options = new DuckDbKeyValueStateStoreOptions(); + private Func _keySerializer; + private Func _valueSerializer; + private Func _keyDeserializer; + private Func _valueDeserializer; + + /// + /// Creates a new builder instance. + /// + /// The name of the state store. + public DuckDbKeyValueStateStoreBuilder(string name) + { + _name = name ?? throw new ArgumentNullException(nameof(name)); + } + + /// + /// Creates a new builder for a DuckDB key-value state store. + /// + /// The name of the state store. + /// A new builder instance. + public static DuckDbKeyValueStateStoreBuilder Create(string name) + { + return new DuckDbKeyValueStateStoreBuilder(name); + } + + /// + /// Configures the store to use an in-memory database. + /// + /// The builder instance for chaining. + public DuckDbKeyValueStateStoreBuilder UseInMemory() + { + _options.UseInMemory = true; + _options.DatabasePath = ":memory:"; + return this; + } + + /// + /// Configures the store to use a persistent database at the specified path. + /// + /// The path to the DuckDB database file. + /// The builder instance for chaining. + public DuckDbKeyValueStateStoreBuilder WithDatabasePath(string databasePath) + { + _options.DatabasePath = databasePath ?? throw new ArgumentNullException(nameof(databasePath)); + _options.UseInMemory = false; + return this; + } + + /// + /// Configures the table name to use for storing key-value pairs. + /// + /// The name of the table. + /// The builder instance for chaining. + public DuckDbKeyValueStateStoreBuilder WithTableName(string tableName) + { + _options.TableName = tableName ?? throw new ArgumentNullException(nameof(tableName)); + return this; + } + + /// + /// Configures whether to create an index on the key column. + /// + /// true to create an index; otherwise, false. + /// The builder instance for chaining. + public DuckDbKeyValueStateStoreBuilder WithIndex(bool createIndex = true) + { + _options.CreateIndex = createIndex; + return this; + } + + /// + /// Configures the maximum memory limit for DuckDB. + /// + /// The maximum memory limit (e.g., "1GB", "512MB"). + /// The builder instance for chaining. + public DuckDbKeyValueStateStoreBuilder WithMaxMemory(string maxMemory) + { + _options.MaxMemory = maxMemory; + return this; + } + + /// + /// Configures the number of threads DuckDB should use. + /// + /// The number of threads. Use 0 for auto-detect. + /// The builder instance for chaining. + public DuckDbKeyValueStateStoreBuilder WithThreads(int threads) + { + _options.Threads = threads; + return this; + } + + /// + /// Configures the access mode for the database. + /// + /// The access mode. + /// The builder instance for chaining. + public DuckDbKeyValueStateStoreBuilder WithAccessMode(DuckDbAccessMode accessMode) + { + _options.AccessMode = accessMode; + return this; + } + + /// + /// Configures a custom key serializer. + /// + /// The key serializer function. + /// The builder instance for chaining. + public DuckDbKeyValueStateStoreBuilder WithKeySerializer(Func serializer) + { + _keySerializer = serializer; + return this; + } + + /// + /// Configures a custom value serializer. + /// + /// The value serializer function. + /// The builder instance for chaining. + public DuckDbKeyValueStateStoreBuilder WithValueSerializer(Func serializer) + { + _valueSerializer = serializer; + return this; + } + + /// + /// Configures a custom key deserializer. + /// + /// The key deserializer function. + /// The builder instance for chaining. + public DuckDbKeyValueStateStoreBuilder WithKeyDeserializer(Func deserializer) + { + _keyDeserializer = deserializer; + return this; + } + + /// + /// Configures a custom value deserializer. + /// + /// The value deserializer function. + /// The builder instance for chaining. + public DuckDbKeyValueStateStoreBuilder WithValueDeserializer(Func deserializer) + { + _valueDeserializer = deserializer; + return this; + } + + /// + /// Builds and returns the configured DuckDB key-value state store. + /// + /// A new DuckDB key-value state store instance. + public DuckDbKeyValueStateStore Build() + { + _options.Validate(); + + return new DuckDbKeyValueStateStore( + _name, + _options, + _keySerializer, + _valueSerializer, + _keyDeserializer, + _valueDeserializer); + } + } +}