diff --git a/.editorconfig b/.editorconfig new file mode 100644 index 0000000..6757b53 --- /dev/null +++ b/.editorconfig @@ -0,0 +1,7 @@ +[*.cs] + +# IDE0066: Convert switch statement to expression +csharp_style_prefer_switch_expression = false + +# IDE0079: Remove unnecessary suppression +dotnet_diagnostic.IDE0079.severity = none diff --git a/.gitignore b/.gitignore index 57d1b93..2a6a2dc 100644 --- a/.gitignore +++ b/.gitignore @@ -431,4 +431,8 @@ FodyWeavers.xsd CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess/Migrations/ # Ignore all scaffolded models and the DbContext from the DbFirst project -CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess.DbFirst/ \ No newline at end of file +CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess.DbFirst/ + +# AI +CLAUDE.md +.claude \ No newline at end of file diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB.Design/Scaffolding/ContinuousAggregateAnnotationApplier.cs b/CmdScale.EntityFrameworkCore.TimescaleDB.Design/Scaffolding/ContinuousAggregateAnnotationApplier.cs new file mode 100644 index 0000000..5cb44f9 --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB.Design/Scaffolding/ContinuousAggregateAnnotationApplier.cs @@ -0,0 +1,35 @@ +using CmdScale.EntityFrameworkCore.TimescaleDB.Configuration.ContinuousAggregate; +using Microsoft.EntityFrameworkCore.Scaffolding.Metadata; +using static CmdScale.EntityFrameworkCore.TimescaleDB.Design.Scaffolding.ContinuousAggregateScaffoldingExtractor; + +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Design.Scaffolding +{ + /// + /// Applies continuous aggregate annotations to scaffolded database views. + /// Note: Continuous aggregates in TimescaleDB are materialized views, so they appear as tables/views in scaffolding. + /// + internal sealed class ContinuousAggregateAnnotationApplier : IAnnotationApplier + { + public void ApplyAnnotations(DatabaseTable table, object featureInfo) + { + if (featureInfo is not ContinuousAggregateInfo info) + { + throw new ArgumentException($"Expected {nameof(ContinuousAggregateInfo)}, got {featureInfo.GetType().Name}", nameof(featureInfo)); + } + + // Mark as a continuous aggregate view + table[ContinuousAggregateAnnotations.MaterializedViewName] = info.MaterializedViewName; + table[ContinuousAggregateAnnotations.ParentName] = info.SourceHypertableName; + table[ContinuousAggregateAnnotations.MaterializedOnly] = info.MaterializedOnly; + + if (!string.IsNullOrEmpty(info.ChunkInterval)) + { + table[ContinuousAggregateAnnotations.ChunkInterval] = info.ChunkInterval; + } + + // Store the view definition for reference (custom annotation) + // This will help users understand the structure when scaffolding + table["TimescaleDB:ViewDefinition"] = info.ViewDefinition; + } + } +} diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB.Design/Scaffolding/ContinuousAggregateScaffoldingExtractor.cs b/CmdScale.EntityFrameworkCore.TimescaleDB.Design/Scaffolding/ContinuousAggregateScaffoldingExtractor.cs new file mode 100644 index 0000000..73e870a --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB.Design/Scaffolding/ContinuousAggregateScaffoldingExtractor.cs @@ -0,0 +1,100 @@ +using System.Data; +using System.Data.Common; + +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Design.Scaffolding +{ + /// + /// Extracts continuous aggregate metadata from a TimescaleDB database for scaffolding. + /// + internal sealed class ContinuousAggregateScaffoldingExtractor : ITimescaleFeatureExtractor + { + internal sealed record ContinuousAggregateInfo( + string MaterializedViewName, + string Schema, + string ViewDefinition, + string SourceHypertableName, + string SourceSchema, + bool MaterializedOnly, + string? ChunkInterval + ); + + public Dictionary<(string Schema, string TableName), object> Extract(DbConnection connection) + { + bool wasOpen = connection.State == ConnectionState.Open; + if (!wasOpen) + { + connection.Open(); + } + + try + { + Dictionary<(string, string), ContinuousAggregateInfo> continuousAggregates = []; + + using (DbCommand command = connection.CreateCommand()) + { + // Query continuous aggregates from TimescaleDB information schema + // This query supports TimescaleDB v2.16 and higher + command.CommandText = @" + SELECT + ca.view_schema, + ca.view_name, + ca.view_definition, + ca.hypertable_schema, + ca.hypertable_name, + ca.materialized_only, + CASE + WHEN d.interval_length IS NOT NULL THEN + (INTERVAL '1 microsecond' * d.interval_length)::text + ELSE NULL + END AS chunk_interval + FROM timescaledb_information.continuous_aggregates ca + LEFT JOIN _timescaledb_catalog.continuous_agg cagg + ON ca.view_schema = cagg.user_view_schema + AND ca.view_name = cagg.user_view_name + LEFT JOIN _timescaledb_catalog.dimension d + ON cagg.mat_hypertable_id = d.hypertable_id + AND d.id = ( + SELECT MIN(d2.id) + FROM _timescaledb_catalog.dimension d2 + WHERE d2.hypertable_id = cagg.mat_hypertable_id + );"; + + using DbDataReader reader = command.ExecuteReader(); + while (reader.Read()) + { + string viewSchema = reader.GetString(0); + string viewName = reader.GetString(1); + string viewDefinition = reader.GetString(2); + string hypertableSchema = reader.GetString(3); + string hypertableName = reader.GetString(4); + bool materializedOnly = reader.GetBoolean(5); + string? chunkInterval = reader.IsDBNull(6) ? null : reader.GetString(6); + + continuousAggregates[(viewSchema, viewName)] = new ContinuousAggregateInfo( + MaterializedViewName: viewName, + Schema: viewSchema, + ViewDefinition: viewDefinition, + SourceHypertableName: hypertableName, + SourceSchema: hypertableSchema, + MaterializedOnly: materializedOnly, + ChunkInterval: chunkInterval + ); + } + } + + // Convert to object dictionary to match interface + return continuousAggregates.ToDictionary( + kvp => kvp.Key, + kvp => (object)kvp.Value + ); + } + finally + { + if (!wasOpen) + { + connection.Close(); + } + } + } + } +} diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB.Design/Scaffolding/HypertableAnnotationApplier.cs b/CmdScale.EntityFrameworkCore.TimescaleDB.Design/Scaffolding/HypertableAnnotationApplier.cs new file mode 100644 index 0000000..a2448f2 --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB.Design/Scaffolding/HypertableAnnotationApplier.cs @@ -0,0 +1,36 @@ +using CmdScale.EntityFrameworkCore.TimescaleDB.Configuration.Hypertable; +using Microsoft.EntityFrameworkCore.Scaffolding.Metadata; +using System.Text.Json; +using static CmdScale.EntityFrameworkCore.TimescaleDB.Design.Scaffolding.HypertableScaffoldingExtractor; + +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Design.Scaffolding +{ + /// + /// Applies hypertable annotations to scaffolded database tables. + /// + internal sealed class HypertableAnnotationApplier : IAnnotationApplier + { + public void ApplyAnnotations(DatabaseTable table, object featureInfo) + { + if (featureInfo is not HypertableInfo info) + { + throw new ArgumentException($"Expected {nameof(HypertableInfo)}, got {featureInfo.GetType().Name}", nameof(featureInfo)); + } + + table[HypertableAnnotations.IsHypertable] = true; + table[HypertableAnnotations.HypertableTimeColumn] = info.TimeColumnName; + table[HypertableAnnotations.ChunkTimeInterval] = info.ChunkTimeInterval; + table[HypertableAnnotations.EnableCompression] = info.CompressionEnabled; + + if (info.ChunkSkipColumns.Count > 0) + { + table[HypertableAnnotations.ChunkSkipColumns] = string.Join(",", info.ChunkSkipColumns); + } + + if (info.AdditionalDimensions.Count > 0) + { + table[HypertableAnnotations.AdditionalDimensions] = JsonSerializer.Serialize(info.AdditionalDimensions); + } + } + } +} diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB.Design/Scaffolding/HypertableScaffoldingExtractor.cs b/CmdScale.EntityFrameworkCore.TimescaleDB.Design/Scaffolding/HypertableScaffoldingExtractor.cs new file mode 100644 index 0000000..e43fe33 --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB.Design/Scaffolding/HypertableScaffoldingExtractor.cs @@ -0,0 +1,156 @@ +using CmdScale.EntityFrameworkCore.TimescaleDB.Abstractions; +using System.Data; +using System.Data.Common; + +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Design.Scaffolding +{ + /// + /// Extracts hypertable metadata from a TimescaleDB database for scaffolding. + /// + internal sealed class HypertableScaffoldingExtractor : ITimescaleFeatureExtractor + { + internal sealed record HypertableInfo( + string TimeColumnName, + string ChunkTimeInterval, + bool CompressionEnabled, + List ChunkSkipColumns, + List AdditionalDimensions + ); + + public Dictionary<(string Schema, string TableName), object> Extract(DbConnection connection) + { + bool wasOpen = connection.State == ConnectionState.Open; + if (!wasOpen) + { + connection.Open(); + } + + try + { + Dictionary<(string, string), HypertableInfo> hypertables = []; + Dictionary<(string, string), bool> compressionSettings = GetCompressionSettings(connection); + + GetHypertableSettings(connection, hypertables, compressionSettings); + GetChunkSkipColumns(connection, hypertables); + + // Convert to object dictionary to match interface + return hypertables.ToDictionary( + kvp => kvp.Key, + kvp => (object)kvp.Value + ); + } + finally + { + if (!wasOpen) + { + connection.Close(); + } + } + } + + private static Dictionary<(string, string), bool> GetCompressionSettings(DbConnection connection) + { + Dictionary<(string, string), bool> compressionSettings = []; + using DbCommand command = connection.CreateCommand(); + command.CommandText = "SELECT hypertable_schema, hypertable_name, compression_enabled FROM timescaledb_information.hypertables;"; + using DbDataReader reader = command.ExecuteReader(); + while (reader.Read()) + { + compressionSettings[(reader.GetString(0), reader.GetString(1))] = reader.GetBoolean(2); + } + return compressionSettings; + } + + private static void GetHypertableSettings( + DbConnection connection, + Dictionary<(string, string), HypertableInfo> hypertables, + Dictionary<(string, string), bool> compressionSettings) + { + using DbCommand command = connection.CreateCommand(); + command.CommandText = @" + SELECT + hypertable_schema, + hypertable_name, + column_name, + dimension_number, + num_partitions, + EXTRACT(EPOCH FROM time_interval) * 1000 AS time_interval_microseconds + FROM timescaledb_information.dimensions + ORDER BY hypertable_schema, hypertable_name, dimension_number;"; + + using DbDataReader reader = command.ExecuteReader(); + while (reader.Read()) + { + string schema = reader.GetString(0); + string name = reader.GetString(1); + string columnName = reader.GetString(2); + int dimensionNumber = reader.GetInt32(3); + + (string schema, string name) key = (schema, name); + + // If it's the first dimension, it defines the primary hypertable settings + if (dimensionNumber == 1) + { + long chunkInterval = reader.IsDBNull(5) ? DefaultValues.ChunkTimeIntervalLong : (long)reader.GetDouble(5); + bool compressionEnabled = compressionSettings.TryGetValue(key, out bool enabled) && enabled; + + hypertables[key] = new HypertableInfo( + TimeColumnName: columnName, + ChunkTimeInterval: chunkInterval.ToString(), + CompressionEnabled: compressionEnabled, + ChunkSkipColumns: [], + AdditionalDimensions: [] + ); + } + // For all other dimensions, add them to the AdditionalDimensions list + else + { + if (hypertables.TryGetValue(key, out HypertableInfo? info)) + { + Dimension dimension; + + if (!reader.IsDBNull(4) && reader.GetInt32(4) > 0) + { + // Space dimension + dimension = Dimension.CreateHash(columnName, reader.GetInt32(4)); + } + else if (!reader.IsDBNull(5)) + { + // Time dimension + long interval = (long)reader.GetDouble(5); + dimension = Dimension.CreateRange(columnName, interval.ToString()); + } + else continue; + + info.AdditionalDimensions.Add(dimension); + } + } + } + } + + private static void GetChunkSkipColumns(DbConnection connection, Dictionary<(string, string), HypertableInfo> hypertables) + { + using DbCommand command = connection.CreateCommand(); + command.CommandText = @" + SELECT + h.schema_name, + h.table_name, + ccs.column_name + FROM _timescaledb_catalog.chunk_column_stats AS ccs + JOIN _timescaledb_catalog.hypertable AS h ON ccs.hypertable_id = h.id;"; + + using DbDataReader reader = command.ExecuteReader(); + while (reader.Read()) + { + string schema = reader.GetString(0); + string name = reader.GetString(1); + string columnName = reader.GetString(2); + + if (hypertables.TryGetValue((schema, name), out HypertableInfo? info)) + { + info.ChunkSkipColumns.Add(columnName); + } + } + } + } +} diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB.Design/Scaffolding/IAnnotationApplier.cs b/CmdScale.EntityFrameworkCore.TimescaleDB.Design/Scaffolding/IAnnotationApplier.cs new file mode 100644 index 0000000..be78345 --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB.Design/Scaffolding/IAnnotationApplier.cs @@ -0,0 +1,15 @@ +using Microsoft.EntityFrameworkCore.Scaffolding.Metadata; + +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Design.Scaffolding +{ + /// + /// Interface for applying TimescaleDB feature annotations to scaffolded database tables. + /// + internal interface IAnnotationApplier + { + /// + /// Applies annotations to the database table based on the feature metadata. + /// + void ApplyAnnotations(DatabaseTable table, object featureInfo); + } +} diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB.Design/Scaffolding/ITimescaleFeatureExtractor.cs b/CmdScale.EntityFrameworkCore.TimescaleDB.Design/Scaffolding/ITimescaleFeatureExtractor.cs new file mode 100644 index 0000000..92a5d23 --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB.Design/Scaffolding/ITimescaleFeatureExtractor.cs @@ -0,0 +1,15 @@ +using System.Data.Common; + +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Design.Scaffolding +{ + /// + /// Interface for extracting TimescaleDB feature metadata from a database connection. + /// + internal interface ITimescaleFeatureExtractor + { + /// + /// Extracts feature metadata from the database and returns a dictionary keyed by (schema, tableName). + /// + Dictionary<(string Schema, string TableName), object> Extract(DbConnection connection); + } +} diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB.Design/Scaffolding/ReorderPolicyAnnotationApplier.cs b/CmdScale.EntityFrameworkCore.TimescaleDB.Design/Scaffolding/ReorderPolicyAnnotationApplier.cs new file mode 100644 index 0000000..fc4c1aa --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB.Design/Scaffolding/ReorderPolicyAnnotationApplier.cs @@ -0,0 +1,49 @@ +using CmdScale.EntityFrameworkCore.TimescaleDB.Configuration.ReorderPolicy; +using Microsoft.EntityFrameworkCore.Scaffolding.Metadata; +using static CmdScale.EntityFrameworkCore.TimescaleDB.Design.Scaffolding.ReorderPolicyScaffoldingExtractor; + +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Design.Scaffolding +{ + /// + /// Applies reorder policy annotations to scaffolded database tables. + /// + internal sealed class ReorderPolicyAnnotationApplier : IAnnotationApplier + { + public void ApplyAnnotations(DatabaseTable table, object featureInfo) + { + if (featureInfo is not ReorderPolicyInfo policyInfo) + { + throw new ArgumentException($"Expected {nameof(ReorderPolicyInfo)}, got {featureInfo.GetType().Name}", nameof(featureInfo)); + } + + table[ReorderPolicyAnnotations.HasReorderPolicy] = true; + table[ReorderPolicyAnnotations.IndexName] = policyInfo.IndexName; + + if (policyInfo.InitialStart.HasValue) + { + table[ReorderPolicyAnnotations.InitialStart] = policyInfo.InitialStart.Value; + } + + // Set annotations only if they differ from TimescaleDB defaults + if (policyInfo.ScheduleInterval != DefaultValues.ReorderPolicyScheduleInterval) + { + table[ReorderPolicyAnnotations.ScheduleInterval] = policyInfo.ScheduleInterval; + } + + if (policyInfo.MaxRuntime != DefaultValues.ReorderPolicyMaxRuntime) + { + table[ReorderPolicyAnnotations.MaxRuntime] = policyInfo.MaxRuntime; + } + + if (policyInfo.MaxRetries != DefaultValues.ReorderPolicyMaxRetries) + { + table[ReorderPolicyAnnotations.MaxRetries] = policyInfo.MaxRetries; + } + + if (policyInfo.RetryPeriod != DefaultValues.ReorderPolicyRetryPeriod) + { + table[ReorderPolicyAnnotations.RetryPeriod] = policyInfo.RetryPeriod; + } + } + } +} diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB.Design/Scaffolding/ReorderPolicyScaffoldingExtractor.cs b/CmdScale.EntityFrameworkCore.TimescaleDB.Design/Scaffolding/ReorderPolicyScaffoldingExtractor.cs new file mode 100644 index 0000000..75b32be --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB.Design/Scaffolding/ReorderPolicyScaffoldingExtractor.cs @@ -0,0 +1,88 @@ +using System.Data; +using System.Data.Common; + +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Design.Scaffolding +{ + /// + /// Extracts reorder policy metadata from a TimescaleDB database for scaffolding. + /// + internal sealed class ReorderPolicyScaffoldingExtractor : ITimescaleFeatureExtractor + { + internal sealed record ReorderPolicyInfo( + string IndexName, + DateTime? InitialStart, + string? ScheduleInterval, + string? MaxRuntime, + int? MaxRetries, + string? RetryPeriod + ); + + public Dictionary<(string Schema, string TableName), object> Extract(DbConnection connection) + { + bool wasOpen = connection.State == ConnectionState.Open; + if (!wasOpen) + { + connection.Open(); + } + + try + { + Dictionary<(string, string), ReorderPolicyInfo> reorderPolicies = []; + using (DbCommand command = connection.CreateCommand()) + { + command.CommandText = @" + SELECT + j.hypertable_schema, + j.hypertable_name, + j.config ->> 'index_name' AS index_name, + j.initial_start, + j.schedule_interval::text, + j.max_runtime::text, + j.max_retries, + j.retry_period::text + FROM timescaledb_information.jobs AS j + WHERE j.proc_name = 'policy_reorder';"; + + using DbDataReader reader = command.ExecuteReader(); + while (reader.Read()) + { + string schema = reader.GetString(0); + string name = reader.GetString(1); + string indexName = reader.GetString(2); + DateTime? initialStart = reader.IsDBNull(3) ? null : reader.GetDateTime(3); + + string? scheduleInterval = reader.IsDBNull(4) ? null : reader.GetString(4); + string? maxRuntime = reader.IsDBNull(5) ? null : reader.GetString(5); + int? maxRetries = reader.IsDBNull(6) ? null : reader.GetInt32(6); + string? retryPeriod = reader.IsDBNull(7) ? null : reader.GetString(7); + + if (!string.IsNullOrEmpty(indexName)) + { + reorderPolicies[(schema, name)] = new ReorderPolicyInfo( + indexName, + initialStart, + scheduleInterval, + maxRuntime, + maxRetries, + retryPeriod + ); + } + } + } + + // Convert to object dictionary to match interface + return reorderPolicies.ToDictionary( + kvp => kvp.Key, + kvp => (object)kvp.Value + ); + } + finally + { + if (!wasOpen) + { + connection.Close(); + } + } + } + } +} diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB.Design/TimescaleCSharpMigrationOperationGenerator.cs b/CmdScale.EntityFrameworkCore.TimescaleDB.Design/TimescaleCSharpMigrationOperationGenerator.cs index ae09e23..dd6160c 100644 --- a/CmdScale.EntityFrameworkCore.TimescaleDB.Design/TimescaleCSharpMigrationOperationGenerator.cs +++ b/CmdScale.EntityFrameworkCore.TimescaleDB.Design/TimescaleCSharpMigrationOperationGenerator.cs @@ -15,7 +15,10 @@ protected override void Generate(MigrationOperation operation, IndentedStringBui HypertableOperationGenerator? hypertableOperationGenerator = null; ReorderPolicyOperationGenerator? reorderPolicyOperationGenerator = null; + ContinuousAggregateOperationGenerator? continuousAggregateOperationGenerator = null; + List statements = []; + bool suppressTransaction = false; switch (operation) { @@ -41,12 +44,26 @@ protected override void Generate(MigrationOperation operation, IndentedStringBui statements = reorderPolicyOperationGenerator.Generate(dropReorder); break; + case CreateContinuousAggregateOperation createContinuousAggregate: + continuousAggregateOperationGenerator ??= new(isDesignTime: true); + statements = continuousAggregateOperationGenerator.Generate(createContinuousAggregate); + suppressTransaction = true; + break; + case AlterContinuousAggregateOperation alterContinuousAggregate: + continuousAggregateOperationGenerator ??= new(isDesignTime: true); + statements = continuousAggregateOperationGenerator.Generate(alterContinuousAggregate); + break; + case DropContinuousAggregateOperation dropContinuousAggregate: + continuousAggregateOperationGenerator ??= new(isDesignTime: true); + statements = continuousAggregateOperationGenerator.Generate(dropContinuousAggregate); + break; + default: base.Generate(operation, builder); break; } - SqlBuilderHelper.BuildQueryString(statements, builder); + SqlBuilderHelper.BuildQueryString(statements, builder, suppressTransaction); } } diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB.Design/TimescaleCSharpModelGenerator.cs b/CmdScale.EntityFrameworkCore.TimescaleDB.Design/TimescaleCSharpModelGenerator.cs deleted file mode 100644 index 1edd468..0000000 --- a/CmdScale.EntityFrameworkCore.TimescaleDB.Design/TimescaleCSharpModelGenerator.cs +++ /dev/null @@ -1,295 +0,0 @@ -using CmdScale.EntityFrameworkCore.TimescaleDB.Abstractions; -using CmdScale.EntityFrameworkCore.TimescaleDB.Configuration.Hypertable; -using CmdScale.EntityFrameworkCore.TimescaleDB.Configuration.ReorderPolicy; -using Microsoft.EntityFrameworkCore; -using Microsoft.EntityFrameworkCore.Diagnostics; -using Microsoft.EntityFrameworkCore.Scaffolding; -using Microsoft.EntityFrameworkCore.Scaffolding.Metadata; -using Npgsql.EntityFrameworkCore.PostgreSQL.Scaffolding.Internal; -using System.Data; -using System.Data.Common; -using System.Text.Json; - -namespace CmdScale.EntityFrameworkCore.TimescaleDB.Design -{ -#pragma warning disable EF1001 - - public class TimescaleDatabaseModelFactory(IDiagnosticsLogger logger) : NpgsqlDatabaseModelFactory(logger) - { - private sealed record HypertableInfo( - string TimeColumnName, - string ChunkTimeInterval, - bool CompressionEnabled, - List ChunkSkipColumns, - List AdditionalDimensions - ); - - private sealed record ReorderPolicyInfo( - string IndexName, - DateTime? InitialStart, - string? ScheduleInterval, - string? MaxRuntime, - int? MaxRetries, - string? RetryPeriod - ); - - public override DatabaseModel Create(DbConnection connection, DatabaseModelFactoryOptions options) - { - DatabaseModel databaseModel = base.Create(connection, options); - Dictionary<(string, string), HypertableInfo> hypertables = GetHypertables(connection); - Dictionary<(string, string), ReorderPolicyInfo> reorderPolicies = GetReorderPolicies(connection); - - // Annotate the tables in the model - foreach (DatabaseTable table in databaseModel.Tables) - { - if (table?.Schema == null) continue; - - (string Schema, string Name) tableKey = (table.Schema, table.Name); - - // Annotations for Hypertables - if (hypertables.TryGetValue(tableKey, out HypertableInfo? info)) - { - table[HypertableAnnotations.IsHypertable] = true; - table[HypertableAnnotations.HypertableTimeColumn] = info.TimeColumnName; - table[HypertableAnnotations.ChunkTimeInterval] = info.ChunkTimeInterval; - table[HypertableAnnotations.EnableCompression] = info.CompressionEnabled; - - if (info.ChunkSkipColumns.Count > 0) - { - table[HypertableAnnotations.ChunkSkipColumns] = string.Join(",", info.ChunkSkipColumns); - } - - if (info.AdditionalDimensions.Count > 0) - { - table[HypertableAnnotations.AdditionalDimensions] = JsonSerializer.Serialize(info.AdditionalDimensions); - } - } - - // Annotate for Reorder Policies - if (reorderPolicies.TryGetValue(tableKey, out ReorderPolicyInfo? policyInfo)) - { - table[ReorderPolicyAnnotations.HasReorderPolicy] = true; - table[ReorderPolicyAnnotations.IndexName] = policyInfo.IndexName; - - if (policyInfo.InitialStart.HasValue) - { - table[ReorderPolicyAnnotations.InitialStart] = policyInfo.InitialStart.Value; - } - - // Set annotations only if they differ from TimescaleDB defaults - if (policyInfo.ScheduleInterval != DefaultValues.ReorderPolicyScheduleInterval) - { - table[ReorderPolicyAnnotations.ScheduleInterval] = policyInfo.ScheduleInterval; - } - - if (policyInfo.MaxRuntime != DefaultValues.ReorderPolicyMaxRuntime) - { - table[ReorderPolicyAnnotations.MaxRuntime] = policyInfo.MaxRuntime; - } - - if (policyInfo.MaxRetries != DefaultValues.ReorderPolicyMaxRetries) - { - table[ReorderPolicyAnnotations.MaxRetries] = policyInfo.MaxRetries; - } - - if (policyInfo.RetryPeriod != DefaultValues.ReorderPolicyRetryPeriod) - { - table[ReorderPolicyAnnotations.RetryPeriod] = policyInfo.RetryPeriod; - } - } - } - - return databaseModel; - } - - private static Dictionary<(string, string), HypertableInfo> GetHypertables(DbConnection connection) - { - bool wasOpen = connection.State == ConnectionState.Open; - if (!wasOpen) - { - connection.Open(); - } - - try - { - Dictionary<(string, string), HypertableInfo> hypertables = []; - Dictionary<(string, string), bool> compressionSettings = []; - - // Get compression settings for hypertables - using (DbCommand command = connection.CreateCommand()) - { - command.CommandText = "SELECT hypertable_schema, hypertable_name, compression_enabled FROM timescaledb_information.hypertables;"; - using DbDataReader reader = command.ExecuteReader(); - while (reader.Read()) - { - compressionSettings[(reader.GetString(0), reader.GetString(1))] = reader.GetBoolean(2); - } - } - - - // Get main hypertable settings - using (DbCommand command = connection.CreateCommand()) - { - command.CommandText = @" - SELECT - hypertable_schema, - hypertable_name, - column_name, - dimension_number, - num_partitions, - EXTRACT(EPOCH FROM time_interval) * 1000 AS time_interval_microseconds - FROM timescaledb_information.dimensions - ORDER BY hypertable_schema, hypertable_name, dimension_number;"; - - using DbDataReader reader = command.ExecuteReader(); - while (reader.Read()) - { - string schema = reader.GetString(0); - string name = reader.GetString(1); - string columnName = reader.GetString(2); - int dimensionNumber = reader.GetInt32(3); - - (string schema, string name) key = (schema, name); - - // If it's the first dimension, it defines the primary hypertable settings - if (dimensionNumber == 1) - { - // long chunkTimeInterval = (long)reader.GetDouble(5); - long chunkInterval = reader.IsDBNull(5) ? DefaultValues.ChunkTimeIntervalLong : (long)reader.GetDouble(5); - bool compressionEnabled = compressionSettings.TryGetValue(key, out bool enabled) && enabled; - - hypertables[key] = new HypertableInfo( - TimeColumnName: columnName, - ChunkTimeInterval: chunkInterval.ToString(), - CompressionEnabled: compressionEnabled, - ChunkSkipColumns: [], - AdditionalDimensions: [] - ); - } - // For all other dimensions, add them to the AdditionalDimensions list - else - { - if (hypertables.TryGetValue(key, out HypertableInfo? info)) - { - Dimension dimension; - - if (!reader.IsDBNull(4) && reader.GetInt32(4) > 0) - { - // Space dimension - dimension = Dimension.CreateHash(columnName, reader.GetInt32(4)); - } - else if (!reader.IsDBNull(5)) - { - // Time dimension - long interval = (long)reader.GetDouble(5); - dimension = Dimension.CreateRange(columnName, interval.ToString()); - } - else continue; - - info.AdditionalDimensions.Add(dimension); - } - } - } - } - - // Get chunk skipping columns and add them to our dictionary - using (DbCommand command = connection.CreateCommand()) - { - command.CommandText = @" - SELECT - h.schema_name, - h.table_name, - ccs.column_name - FROM _timescaledb_catalog.chunk_column_stats AS ccs - JOIN _timescaledb_catalog.hypertable AS h ON ccs.hypertable_id = h.id;"; - - using DbDataReader reader = command.ExecuteReader(); - while (reader.Read()) - { - string schema = reader.GetString(0); - string name = reader.GetString(1); - string columnName = reader.GetString(2); - - if (hypertables.TryGetValue((schema, name), out HypertableInfo? info)) - { - info.ChunkSkipColumns.Add(columnName); - } - } - } - - return hypertables; - } - finally - { - if (!wasOpen) - { - connection.Close(); - } - } - } - - private static Dictionary<(string, string), ReorderPolicyInfo> GetReorderPolicies(DbConnection connection) - { - bool wasOpen = connection.State == ConnectionState.Open; - if (!wasOpen) - { - connection.Open(); - } - - try - { - Dictionary<(string, string), ReorderPolicyInfo> reorderPolicies = []; - using (DbCommand command = connection.CreateCommand()) - { - command.CommandText = @" - SELECT - j.hypertable_schema, - j.hypertable_name, - j.config ->> 'index_name' AS index_name, - j.initial_start, - j.schedule_interval::text, - j.max_runtime::text, - j.max_retries, - j.retry_period::text - FROM timescaledb_information.jobs AS j - WHERE j.proc_name = 'policy_reorder';"; - - using DbDataReader reader = command.ExecuteReader(); - while (reader.Read()) - { - string schema = reader.GetString(0); - string name = reader.GetString(1); - string indexName = reader.GetString(2); - DateTime? initialStart = reader.IsDBNull(3) ? null : reader.GetDateTime(3); - - string? scheduleInterval = reader.IsDBNull(4) ? null : reader.GetString(4); - string? maxRuntime = reader.IsDBNull(5) ? null : reader.GetString(5); - int? maxRetries = reader.IsDBNull(6) ? null : reader.GetInt32(6); - string? retryPeriod = reader.IsDBNull(7) ? null : reader.GetString(7); - - if (!string.IsNullOrEmpty(indexName)) - { - reorderPolicies[(schema, name)] = new ReorderPolicyInfo( - indexName, - initialStart, - scheduleInterval, - maxRuntime, - maxRetries, - retryPeriod - ); - } - } - } - return reorderPolicies; - } - finally - { - if (!wasOpen) - { - connection.Close(); - } - } - } - } -#pragma warning restore EF1001 -} diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB.Design/TimescaleDatabaseModelFactory.cs b/CmdScale.EntityFrameworkCore.TimescaleDB.Design/TimescaleDatabaseModelFactory.cs new file mode 100644 index 0000000..86f833e --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB.Design/TimescaleDatabaseModelFactory.cs @@ -0,0 +1,57 @@ +using CmdScale.EntityFrameworkCore.TimescaleDB.Design.Scaffolding; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Diagnostics; +using Microsoft.EntityFrameworkCore.Scaffolding; +using Microsoft.EntityFrameworkCore.Scaffolding.Metadata; +using Npgsql.EntityFrameworkCore.PostgreSQL.Scaffolding.Internal; +using System.Data.Common; + +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Design +{ +#pragma warning disable EF1001 + /// + /// Database model factory that extends Npgsql's scaffolding to include TimescaleDB-specific features. + /// Handles extraction of hypertables, reorder policies, and continuous aggregates during db-first scaffolding. + /// + public class TimescaleDatabaseModelFactory(IDiagnosticsLogger logger) + : NpgsqlDatabaseModelFactory(logger) + { + private readonly List<(ITimescaleFeatureExtractor Extractor, IAnnotationApplier Applier)> _features = + [ + (new HypertableScaffoldingExtractor(), new HypertableAnnotationApplier()), + (new ReorderPolicyScaffoldingExtractor(), new ReorderPolicyAnnotationApplier()), + (new ContinuousAggregateScaffoldingExtractor(), new ContinuousAggregateAnnotationApplier()) + ]; + + public override DatabaseModel Create(DbConnection connection, DatabaseModelFactoryOptions options) + { + DatabaseModel databaseModel = base.Create(connection, options); + + // Extract all TimescaleDB features from the database + var allFeatureData = _features + .Select(feature => feature.Extractor.Extract(connection)) + .ToList(); + + // Apply annotations to tables/views in the model + foreach (DatabaseTable table in databaseModel.Tables) + { + if (table?.Schema == null) continue; + + (string Schema, string Name) tableKey = (table.Schema, table.Name); + + // Apply each feature's annotations if the table has that feature + for (int i = 0; i < _features.Count; i++) + { + var featureData = allFeatureData[i]; + if (featureData.TryGetValue(tableKey, out object? featureInfo)) + { + _features[i].Applier.ApplyAnnotations(table, featureInfo); + } + } + } + + return databaseModel; + } + } +#pragma warning restore EF1001 +} diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess/Configurations/TradeAggregateConfiguration.cs b/CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess/Configurations/TradeAggregateConfiguration.cs new file mode 100644 index 0000000..040ed6b --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess/Configurations/TradeAggregateConfiguration.cs @@ -0,0 +1,24 @@ +using CmdScale.EntityFrameworkCore.TimescaleDB.Abstractions; +using CmdScale.EntityFrameworkCore.TimescaleDB.Configuration.ContinuousAggregate; +using CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess.Models; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Metadata.Builders; + +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess.Configurations +{ + public class TradeAggregateConfiguration : IEntityTypeConfiguration + { + public void Configure(EntityTypeBuilder builder) + { + builder.HasNoKey(); + builder.IsContinuousAggregate("trade_aggregate_view", "1 hour", x => x.Timestamp, true, "7 days") + .AddAggregateFunction(x => x.AveragePrice, x => x.Price, EAggregateFunction.Avg) + .AddAggregateFunction(x => x.MinPrice, x => x.Price, EAggregateFunction.Max) + .AddAggregateFunction(x => x.MaxPrice, x => x.Price, EAggregateFunction.Min) + .AddGroupByColumn(x => x.Exchange) + .AddGroupByColumn("1, 2") + .Where("\"ticker\" = 'MCRS'") + .MaterializedOnly(); + } + } +} diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess/Configurations/TradeConfiguration.cs b/CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess/Configurations/TradeConfiguration.cs index d645ce3..8d8ee85 100644 --- a/CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess/Configurations/TradeConfiguration.cs +++ b/CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess/Configurations/TradeConfiguration.cs @@ -14,7 +14,8 @@ public void Configure(EntityTypeBuilder builder) builder.HasNoKey() .IsHypertable(x => x.Timestamp) .WithChunkTimeInterval("1 day"); - builder.WithReorderPolicy("Trades_Timestamp_idx", DateTime.Parse("2025-09-23T09:15:19.3905112Z"), "2 days", "10 minutes", null, "1 minute"); + builder.HasIndex(x => x.Timestamp).HasDatabaseName("Trades_Timestamp_idx"); + builder.WithReorderPolicy("Trades_Timestamp_idx", DateTime.Parse("2025-09-23T09:15:19.3905112Z"), "2 days", "10 minutes", -1, "1 minute"); } } } diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess/Models/DeviceReading.cs b/CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess/Models/DeviceReading.cs index 12bb12b..5b6affb 100644 --- a/CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess/Models/DeviceReading.cs +++ b/CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess/Models/DeviceReading.cs @@ -4,9 +4,10 @@ namespace CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess.Models { - [Hypertable(nameof(Time), ChunkSkipColumns = new[] { "Time" }, ChunkTimeInterval = "1 day")] - [ReorderPolicy("DeviceReadings_Time_idx", InitialStart = "2025-09-23T09:15:19.3905112Z", ScheduleInterval = "1 day", MaxRuntime = "00:00:00", RetryPeriod = "00:05:00")] + [Hypertable(nameof(Time), ChunkSkipColumns = new[] { "Time" }, ChunkTimeInterval = "1 day", EnableCompression = true)] + [Index(nameof(Time), Name = "ix_device_readings_time")] [PrimaryKey(nameof(Id), nameof(Time))] + [ReorderPolicy("ix_device_readings_time", InitialStart = "2025-09-23T09:15:19.3905112Z", ScheduleInterval = "1 day", MaxRuntime = "00:00:00", RetryPeriod = "00:05:00", MaxRetries = 3)] public class DeviceReading { public Guid Id { get; set; } diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess/Models/TradeAggregate.cs b/CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess/Models/TradeAggregate.cs new file mode 100644 index 0000000..dfc76fa --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess/Models/TradeAggregate.cs @@ -0,0 +1,9 @@ +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess.Models +{ + public class TradeAggregate + { + public decimal AveragePrice { get; set; } + public decimal MaxPrice { get; set; } + public decimal MinPrice { get; set; } + } +} diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess/Models/WeatherAggregate.cs b/CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess/Models/WeatherAggregate.cs new file mode 100644 index 0000000..8072d4d --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess/Models/WeatherAggregate.cs @@ -0,0 +1,51 @@ +using CmdScale.EntityFrameworkCore.TimescaleDB.Abstractions; +using CmdScale.EntityFrameworkCore.TimescaleDB.Configuration.ContinuousAggregate; +using Microsoft.EntityFrameworkCore; + +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess.Models +{ + /// + /// Example continuous aggregate showcasing all possible configuration properties and aggregate functions. + /// This aggregates weather data into daily buckets with various statistical measures. + /// + [Keyless] + [ContinuousAggregate( + MaterializedViewName = "weather_aggregates", + ParentName = nameof(WeatherData), + ChunkInterval = "1 month", + WithNoData = true, + CreateGroupIndexes = true, + MaterializedOnly = false, + Where = "\"temperature\" > -50 AND \"humidity\" >= 0")] + [TimeBucket("1 day", nameof(WeatherData.Time), GroupBy = true)] + public class WeatherAggregate + { + // Avg aggregate function + [Aggregate(EAggregateFunction.Avg, nameof(WeatherData.Temperature))] + public double AverageTemperature { get; set; } + + // Max aggregate function + [Aggregate(EAggregateFunction.Max, nameof(WeatherData.Humidity))] + public double MaxHumidity { get; set; } + + // Min aggregate function + [Aggregate(EAggregateFunction.Min, nameof(WeatherData.Humidity))] + public double MinHumidity { get; set; } + + // Sum aggregate function + [Aggregate(EAggregateFunction.Sum, nameof(WeatherData.Temperature))] + public double TotalTemperature { get; set; } + + // Count aggregate function (using "*" for count all records) + [Aggregate(EAggregateFunction.Count, "*")] + public int RecordCount { get; set; } + + // First aggregate function (gets first temperature value in time bucket) + [Aggregate(EAggregateFunction.First, nameof(WeatherData.Temperature))] + public double FirstTemperature { get; set; } + + // Last aggregate function (gets last temperature value in time bucket) + [Aggregate(EAggregateFunction.Last, nameof(WeatherData.Temperature))] + public double LastTemperature { get; set; } + } +} diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess/TimescaleContext.cs b/CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess/TimescaleContext.cs index e7bd612..84e92dd 100644 --- a/CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess/TimescaleContext.cs +++ b/CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess/TimescaleContext.cs @@ -11,6 +11,8 @@ public class TimescaleContext(DbContextOptions options) : DbCo public DbSet OrderStatusEvents { get; set; } public DbSet Trades { get; set; } public DbSet TradesWithId { get; set; } + public DbSet TradeAggregates { get; set; } + public DbSet WeatherAggregates { get; set; } protected override void OnModelCreating(ModelBuilder modelBuilder) { diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB.Tests/CmdScale.EntityFrameworkCore.TimescaleDB.Tests.csproj b/CmdScale.EntityFrameworkCore.TimescaleDB.Tests/CmdScale.EntityFrameworkCore.TimescaleDB.Tests.csproj index d9fac6c..aa7afc4 100644 --- a/CmdScale.EntityFrameworkCore.TimescaleDB.Tests/CmdScale.EntityFrameworkCore.TimescaleDB.Tests.csproj +++ b/CmdScale.EntityFrameworkCore.TimescaleDB.Tests/CmdScale.EntityFrameworkCore.TimescaleDB.Tests.csproj @@ -14,8 +14,14 @@ all runtime; build; native; contentfiles; analyzers; buildtransitive + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + all diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB.Tests/Integration/ContinuousAggregateIntegrationTests.cs b/CmdScale.EntityFrameworkCore.TimescaleDB.Tests/Integration/ContinuousAggregateIntegrationTests.cs new file mode 100644 index 0000000..5e3df5f --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB.Tests/Integration/ContinuousAggregateIntegrationTests.cs @@ -0,0 +1,1128 @@ +using CmdScale.EntityFrameworkCore.TimescaleDB.Abstractions; +using CmdScale.EntityFrameworkCore.TimescaleDB.Configuration.ContinuousAggregate; +using CmdScale.EntityFrameworkCore.TimescaleDB.Configuration.Hypertable; +using Microsoft.EntityFrameworkCore; +using Testcontainers.PostgreSql; + +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Tests.Integration +{ + public class ContinuousAggregateIntegrationTests : IAsyncLifetime + { + private PostgreSqlContainer? _container; + private string? _connectionString; + + public async Task InitializeAsync() + { + // Arrange: Start TimescaleDB container + _container = new PostgreSqlBuilder() + .WithImage("timescale/timescaledb:latest-pg16") + .WithDatabase("test_db") + .WithUsername("test_user") + .WithPassword("test_password") + .Build(); + + await _container.StartAsync(); + _connectionString = _container.GetConnectionString(); + } + + public async Task DisposeAsync() + { + if (_container != null) + { + await _container.DisposeAsync(); + } + } + + [Fact] + public async Task Should_Create_ContinuousAggregate_With_BasicAggregates() + { + // Arrange: Create context with hypertable and continuous aggregate using basic aggregates + await using var context = new BasicAggregatesTestContext(_connectionString!); + await context.Database.EnsureCreatedAsync(); + + // Insert test data + await InsertTradeDataAsync(context); + + // Act: Refresh the continuous aggregate + await context.Database.ExecuteSqlRawAsync( + "CALL refresh_continuous_aggregate('public.trade_aggregate_basic', NULL, NULL);"); + + var aggregates = await context.TradeAggregates + .OrderBy(a => a.TimeBucket) + .ToListAsync(); + + // Assert: Verify aggregates were calculated correctly + Assert.NotEmpty(aggregates); + var firstAggregate = aggregates.First(); + Assert.True(firstAggregate.AvgPrice > 0); + Assert.True(firstAggregate.MaxPrice >= firstAggregate.MinPrice); + Assert.True(firstAggregate.SumPrice > 0); + Assert.True(firstAggregate.CountPrice > 0); + } + + [Fact] + public async Task Should_Create_ContinuousAggregate_With_FirstAndLast_Functions() + { + // Arrange: Create context with First and Last aggregate functions + await using var context = new FirstLastTestContext(_connectionString!); + await context.Database.EnsureCreatedAsync(); + + // Insert test data with specific timestamps + await context.Database.ExecuteSqlRawAsync(@" + INSERT INTO ""Trades"" (""Timestamp"", ""Ticker"", ""Price"", ""Size"", ""Exchange"") + VALUES + ('2025-01-06 10:00:00+00', 'AAPL', 100.00, 100, 'NYSE'), + ('2025-01-06 10:30:00+00', 'AAPL', 105.00, 200, 'NYSE'), + ('2025-01-06 10:45:00+00', 'AAPL', 103.00, 150, 'NYSE'); + "); + + // Act: Refresh the continuous aggregate + await context.Database.ExecuteSqlRawAsync( + "CALL refresh_continuous_aggregate('public.trade_aggregate_first_last', NULL, NULL);"); + + var aggregates = await context.TradeAggregates.ToListAsync(); + + // Assert: Verify first() returns earliest value and last() returns latest value + Assert.Single(aggregates); + Assert.Equal(100.00m, aggregates[0].FirstPrice); // First price at 10:00 + Assert.Equal(103.00m, aggregates[0].LastPrice); // Last price at 10:45 + } + + [Fact] + public async Task Should_Create_ContinuousAggregate_With_GroupByColumns() + { + // Arrange: Create context with GROUP BY columns + await using var context = new GroupByTestContext(_connectionString!); + await context.Database.EnsureCreatedAsync(); + + // Insert test data for different exchanges + await context.Database.ExecuteSqlRawAsync(@" + INSERT INTO ""Trades"" (""Timestamp"", ""Ticker"", ""Price"", ""Size"", ""Exchange"") + VALUES + ('2025-01-06 10:00:00+00', 'AAPL', 100.00, 100, 'NYSE'), + ('2025-01-06 10:00:00+00', 'AAPL', 110.00, 200, 'NASDAQ'), + ('2025-01-06 10:00:00+00', 'AAPL', 105.00, 150, 'LSE'); + "); + + // Act: Refresh the continuous aggregate + await context.Database.ExecuteSqlRawAsync( + "CALL refresh_continuous_aggregate('public.trade_aggregate_grouped', NULL, NULL);"); + + var aggregates = await context.TradeAggregates + .OrderBy(a => a.Exchange) + .ToListAsync(); + + // Assert: Verify we have one aggregate per exchange + Assert.Equal(3, aggregates.Count); + Assert.Equal("LSE", aggregates[0].Exchange); + Assert.Equal(105.00m, aggregates[0].AvgPrice); + Assert.Equal("NASDAQ", aggregates[1].Exchange); + Assert.Equal(110.00m, aggregates[1].AvgPrice); + Assert.Equal("NYSE", aggregates[2].Exchange); + Assert.Equal(100.00m, aggregates[2].AvgPrice); + } + + [Fact] + public async Task Should_Create_ContinuousAggregate_With_WhereClause() + { + // Arrange: Create context with WHERE clause to filter data + await using var context = new WhereClauseTestContext(_connectionString!); + await context.Database.EnsureCreatedAsync(); + + // Insert test data with different tickers + await context.Database.ExecuteSqlRawAsync(@" + INSERT INTO ""Trades"" (""Timestamp"", ""Ticker"", ""Price"", ""Size"", ""Exchange"") + VALUES + ('2025-01-06 10:00:00+00', 'AAPL', 100.00, 100, 'NYSE'), + ('2025-01-06 10:00:00+00', 'TSLA', 200.00, 200, 'NYSE'), + ('2025-01-06 10:00:00+00', 'MSFT', 300.00, 150, 'NYSE'); + "); + + // Act: Refresh the continuous aggregate (should only include AAPL) + await context.Database.ExecuteSqlRawAsync( + "CALL refresh_continuous_aggregate('public.trade_aggregate_filtered', NULL, NULL);"); + + var aggregates = await context.TradeAggregates.ToListAsync(); + + // Assert: Verify only AAPL data is included + Assert.Single(aggregates); + Assert.Equal(100.00m, aggregates[0].AvgPrice); + } + + [Fact] + public async Task Should_Create_ContinuousAggregate_WithNoData_Option() + { + // Arrange: Create context with WITH NO DATA option + await using var context = new WithNoDataTestContext(_connectionString!); + await context.Database.EnsureCreatedAsync(); + + // Insert test data + await context.Database.ExecuteSqlRawAsync(@" + INSERT INTO ""Trades"" (""Timestamp"", ""Ticker"", ""Price"", ""Size"", ""Exchange"") + VALUES + ('2025-01-06 10:00:00+00', 'AAPL', 100.00, 100, 'NYSE'); + "); + + // Act: Query the continuous aggregate (should be empty because of WITH NO DATA) + var aggregates = await context.TradeAggregates.ToListAsync(); + + // Assert: Verify no data is materialized initially + Assert.Empty(aggregates); + + // Now refresh and verify data appears + await context.Database.ExecuteSqlRawAsync( + "CALL refresh_continuous_aggregate('public.trade_aggregate_no_data', NULL, NULL);"); + + aggregates = await context.TradeAggregates.ToListAsync(); + Assert.Single(aggregates); + } + + [Fact] + public async Task Should_Create_ContinuousAggregate_With_CustomChunkInterval() + { + // Arrange: Create context with custom chunk_interval = "1 day" + await using var context = new CustomChunkIntervalTestContext(_connectionString!); + await context.Database.EnsureCreatedAsync(); + + // Insert test data + await InsertTradeDataAsync(context); + + // Act: Refresh and query the aggregate + await context.Database.ExecuteSqlRawAsync( + "CALL refresh_continuous_aggregate('public.trade_aggregate_custom_chunk', NULL, NULL);"); + + var aggregates = await context.TradeAggregates.ToListAsync(); + + // Assert: Verify the continuous aggregate works correctly + Assert.NotEmpty(aggregates); + } + + [Fact] + public async Task Should_Create_ContinuousAggregate_With_CreateGroupIndexes() + { + // Arrange: Create context with create_group_indexes = true + await using var context = new CreateGroupIndexesTestContext(_connectionString!); + await context.Database.EnsureCreatedAsync(); + + // Insert test data + await InsertTradeDataAsync(context); + + // Act: Refresh and query the aggregate + await context.Database.ExecuteSqlRawAsync( + "CALL refresh_continuous_aggregate('public.trade_aggregate_with_indexes', NULL, NULL);"); + + var aggregates = await context.TradeAggregates.ToListAsync(); + + // Assert: Verify aggregates were created (indexes are internal, hard to verify directly) + Assert.NotEmpty(aggregates); + } + + [Fact] + public async Task Should_Create_ContinuousAggregate_With_MaterializedOnly_False() + { + // Arrange: Create context with materialized_only = false (allows real-time aggregation) + await using var context = new MaterializedOnlyFalseTestContext(_connectionString!); + await context.Database.EnsureCreatedAsync(); + + // Insert test data + await context.Database.ExecuteSqlRawAsync(@" + INSERT INTO ""Trades"" (""Timestamp"", ""Ticker"", ""Price"", ""Size"", ""Exchange"") + VALUES + ('2025-01-06 10:00:00+00', 'AAPL', 100.00, 100, 'NYSE'); + "); + + // Act: Query without explicit refresh (should include real-time data) + var aggregates = await context.TradeAggregates.ToListAsync(); + + // Assert: Verify we can see data even without manual refresh + Assert.Single(aggregates); + Assert.Equal(100.00m, aggregates[0].AvgPrice); + } + + [Fact] + public async Task Should_Alter_ContinuousAggregate_ChunkInterval() + { + // Arrange: Create context with initial chunk_interval = "7 days" + await using var context1 = new AlterChunkIntervalContext_Before(_connectionString!); + await context1.Database.EnsureCreatedAsync(); + + // Insert test data and refresh + await InsertTradeDataAsync(context1); + await context1.Database.ExecuteSqlRawAsync( + "CALL refresh_continuous_aggregate('public.trade_aggregate_alterable', NULL, NULL);"); + + var aggregatesBefore = await context1.TradeAggregates.ToListAsync(); + Assert.NotEmpty(aggregatesBefore); + + // Act: Alter the chunk_interval to "14 days" + await using var context2 = new AlterChunkIntervalContext_After(_connectionString!); + + await context2.Database.ExecuteSqlRawAsync(@" + ALTER MATERIALIZED VIEW trade_aggregate_alterable + SET (timescaledb.chunk_interval = '14 days'); + "); + + // Assert: Verify we can still query the aggregate after altering chunk_interval + var aggregatesAfter = await context2.TradeAggregates.ToListAsync(); + Assert.NotEmpty(aggregatesAfter); + Assert.Equal(aggregatesBefore.Count, aggregatesAfter.Count); + } + + [Fact] + public async Task Should_Alter_ContinuousAggregate_MaterializedOnly() + { + // Arrange: Create context with materialized_only = false + await using var context = new AlterMaterializedOnlyTestContext(_connectionString!); + await context.Database.EnsureCreatedAsync(); + + // Insert test data and refresh + await InsertTradeDataAsync(context); + await context.Database.ExecuteSqlRawAsync( + "CALL refresh_continuous_aggregate('public.trade_aggregate_materialized_only', NULL, NULL);"); + + // Act: Alter materialized_only to true + await context.Database.ExecuteSqlRawAsync(@" + ALTER MATERIALIZED VIEW trade_aggregate_materialized_only + SET (timescaledb.materialized_only = true); + "); + + // Assert: Verify we can still query the aggregate after alteration + var aggregates = await context.TradeAggregates.ToListAsync(); + Assert.NotEmpty(aggregates); + } + + [Fact] + public async Task Should_Alter_ContinuousAggregate_CreateGroupIndexes() + { + // NOTE: TimescaleDB does not support altering create_group_indexes after creation + // This test verifies that the option is set during creation but cannot be altered + + // Arrange: Create context with create_group_indexes = false + await using var context = new AlterCreateGroupIndexesTestContext(_connectionString!); + await context.Database.EnsureCreatedAsync(); + + // Act & Assert: Attempting to alter create_group_indexes should fail + await Assert.ThrowsAsync(async () => + { + await context.Database.ExecuteSqlRawAsync(@" + ALTER MATERIALIZED VIEW trade_aggregate_group_indexes + SET (timescaledb.create_group_indexes = true); + "); + }); + } + + [Fact] + public async Task Should_Drop_ContinuousAggregate_Successfully() + { + // Arrange: Create context with continuous aggregate + await using var context = new DropTestContext(_connectionString!); + await context.Database.EnsureCreatedAsync(); + + // Insert data and refresh to ensure aggregate has data + await InsertTradeDataAsync(context); + await context.Database.ExecuteSqlRawAsync( + "CALL refresh_continuous_aggregate('public.trade_aggregate_to_drop', NULL, NULL);"); + + // Verify we can query the aggregate before dropping + var aggregatesBefore = await context.TradeAggregates.ToListAsync(); + Assert.NotEmpty(aggregatesBefore); + + // Act: Drop the continuous aggregate + await context.Database.ExecuteSqlRawAsync( + "DROP MATERIALIZED VIEW IF EXISTS trade_aggregate_to_drop;"); + + // Assert: Verify we cannot query the aggregate after dropping (should throw) + await Assert.ThrowsAsync(async () => + { + await context.TradeAggregates.ToListAsync(); + }); + } + + [Fact] + public async Task Should_Generate_Correct_SQL_For_ContinuousAggregate() + { + // Arrange: Create context and ensure database is created + await using var context = new SqlGenerationTestContext(_connectionString!); + await context.Database.EnsureCreatedAsync(); + + // Insert test data + await InsertTradeDataAsync(context); + + // Act: Refresh and query the continuous aggregate + await context.Database.ExecuteSqlRawAsync( + "CALL refresh_continuous_aggregate('public.trade_aggregate_sql_gen', NULL, NULL);"); + + var aggregates = await context.TradeAggregates.ToListAsync(); + + // Assert: Verify the continuous aggregate works correctly + Assert.NotEmpty(aggregates); + var firstAggregate = aggregates.First(); + Assert.True(firstAggregate.AvgPrice > 0); + } + + [Fact] + public async Task Should_Handle_SnakeCase_Naming_Convention() + { + // Arrange: Create context with snake_case naming convention + await using var context = new SnakeCaseTestContext(_connectionString!); + await context.Database.EnsureCreatedAsync(); + + // Insert test data + await context.Database.ExecuteSqlRawAsync(@" + INSERT INTO trades (timestamp, ticker, price, size, exchange) + VALUES + ('2025-01-06 10:00:00+00', 'AAPL', 100.00, 100, 'NYSE'); + "); + + // Act: Refresh and query the continuous aggregate + await context.Database.ExecuteSqlRawAsync( + "CALL refresh_continuous_aggregate('public.trade_aggregate_snake_case', NULL, NULL);"); + + var aggregates = await context.TradeAggregates.ToListAsync(); + + // Assert: Verify snake_case columns work correctly + Assert.Single(aggregates); + Assert.Equal(100.00m, aggregates[0].avg_price); + } + + #region Helper Methods + + private async Task InsertTradeDataAsync(DbContext context) + { + await context.Database.ExecuteSqlRawAsync(@" + INSERT INTO ""Trades"" (""Timestamp"", ""Ticker"", ""Price"", ""Size"", ""Exchange"") + VALUES + ('2025-01-06 10:00:00+00', 'AAPL', 150.50, 100, 'NYSE'), + ('2025-01-06 10:30:00+00', 'AAPL', 151.00, 200, 'NYSE'), + ('2025-01-06 10:45:00+00', 'AAPL', 149.75, 150, 'NYSE'); + "); + } + + #endregion + + #region Test Models + + private class TestTrade + { + public DateTime Timestamp { get; set; } + public string Ticker { get; set; } = string.Empty; + public decimal Price { get; set; } + public int Size { get; set; } + public string Exchange { get; set; } = string.Empty; + } + + private class BasicAggregatesTestAggregate + { + public DateTime TimeBucket { get; set; } + public decimal AvgPrice { get; set; } + public decimal MaxPrice { get; set; } + public decimal MinPrice { get; set; } + public decimal SumPrice { get; set; } + public long CountPrice { get; set; } + } + + private class FirstLastTestAggregate + { + public DateTime TimeBucket { get; set; } + public decimal FirstPrice { get; set; } + public decimal LastPrice { get; set; } + } + + private class GroupByTestAggregate + { + public DateTime TimeBucket { get; set; } + public string Exchange { get; set; } = string.Empty; + public decimal AvgPrice { get; set; } + } + + private class WhereClauseTestAggregate + { + public DateTime TimeBucket { get; set; } + public decimal AvgPrice { get; set; } + } + + private class SnakeCaseTestAggregate + { + public DateTime time_bucket { get; set; } + public decimal avg_price { get; set; } + } + + #endregion + + #region Test Contexts + + private class BasicAggregatesTestContext : DbContext + { + private readonly string _connectionString; + + public BasicAggregatesTestContext(string connectionString) + { + _connectionString = connectionString; + } + + public DbSet Trades => Set(); + public DbSet TradeAggregates => Set(); + + protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder) + { + optionsBuilder.UseNpgsql(_connectionString) + .UseTimescaleDb(); + } + + protected override void OnModelCreating(ModelBuilder modelBuilder) + { + // Configure Trade as a hypertable + modelBuilder.Entity(entity => + { + entity.ToTable("Trades"); + entity.HasNoKey(); + entity.IsHypertable(x => x.Timestamp); + }); + + // Configure continuous aggregate with all basic aggregate functions + modelBuilder.Entity(entity => + { + entity.HasNoKey(); + entity.IsContinuousAggregate( + "trade_aggregate_basic", + "1 hour", + x => x.Timestamp) + .AddAggregateFunction(x => x.AvgPrice, x => x.Price, EAggregateFunction.Avg) + .AddAggregateFunction(x => x.MaxPrice, x => x.Price, EAggregateFunction.Max) + .AddAggregateFunction(x => x.MinPrice, x => x.Price, EAggregateFunction.Min) + .AddAggregateFunction(x => x.SumPrice, x => x.Price, EAggregateFunction.Sum) + .AddAggregateFunction(x => x.CountPrice, x => x.Price, EAggregateFunction.Count); + + // Map properties to view columns + entity.Property(x => x.TimeBucket).HasColumnName("time_bucket"); + entity.Property(x => x.AvgPrice).HasColumnName("AvgPrice"); + entity.Property(x => x.MaxPrice).HasColumnName("MaxPrice"); + entity.Property(x => x.MinPrice).HasColumnName("MinPrice"); + entity.Property(x => x.SumPrice).HasColumnName("SumPrice"); + entity.Property(x => x.CountPrice).HasColumnName("CountPrice"); + }); + } + } + + private class FirstLastTestContext : DbContext + { + private readonly string _connectionString; + + public FirstLastTestContext(string connectionString) + { + _connectionString = connectionString; + } + + public DbSet Trades => Set(); + public DbSet TradeAggregates => Set(); + + protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder) + { + optionsBuilder.UseNpgsql(_connectionString) + .UseTimescaleDb(); + } + + protected override void OnModelCreating(ModelBuilder modelBuilder) + { + modelBuilder.Entity(entity => + { + entity.ToTable("Trades"); + entity.HasNoKey(); + entity.IsHypertable(x => x.Timestamp); + }); + + modelBuilder.Entity(entity => + { + entity.HasNoKey(); + entity.IsContinuousAggregate( + "trade_aggregate_first_last", + "1 hour", + x => x.Timestamp) + .AddAggregateFunction(x => x.FirstPrice, x => x.Price, EAggregateFunction.First) + .AddAggregateFunction(x => x.LastPrice, x => x.Price, EAggregateFunction.Last); + + // Map properties to view columns + entity.Property(x => x.TimeBucket).HasColumnName("time_bucket"); + entity.Property(x => x.FirstPrice).HasColumnName("FirstPrice"); + entity.Property(x => x.LastPrice).HasColumnName("LastPrice"); + }); + } + } + + private class GroupByTestContext : DbContext + { + private readonly string _connectionString; + + public GroupByTestContext(string connectionString) + { + _connectionString = connectionString; + } + + public DbSet Trades => Set(); + public DbSet TradeAggregates => Set(); + + protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder) + { + optionsBuilder.UseNpgsql(_connectionString) + .UseTimescaleDb(); + } + + protected override void OnModelCreating(ModelBuilder modelBuilder) + { + modelBuilder.Entity(entity => + { + entity.ToTable("Trades"); + entity.HasNoKey(); + entity.IsHypertable(x => x.Timestamp); + }); + + modelBuilder.Entity(entity => + { + entity.HasNoKey(); + entity.IsContinuousAggregate( + "trade_aggregate_grouped", + "1 hour", + x => x.Timestamp) + .AddAggregateFunction(x => x.AvgPrice, x => x.Price, EAggregateFunction.Avg) + .AddGroupByColumn(x => x.Exchange); + + // Map properties to view columns + entity.Property(x => x.TimeBucket).HasColumnName("time_bucket"); + entity.Property(x => x.Exchange).HasColumnName("Exchange"); + entity.Property(x => x.AvgPrice).HasColumnName("AvgPrice"); + }); + } + } + + private class WhereClauseTestContext : DbContext + { + private readonly string _connectionString; + + public WhereClauseTestContext(string connectionString) + { + _connectionString = connectionString; + } + + public DbSet Trades => Set(); + public DbSet TradeAggregates => Set(); + + protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder) + { + optionsBuilder.UseNpgsql(_connectionString) + .UseTimescaleDb(); + } + + protected override void OnModelCreating(ModelBuilder modelBuilder) + { + modelBuilder.Entity(entity => + { + entity.ToTable("Trades"); + entity.HasNoKey(); + entity.IsHypertable(x => x.Timestamp); + }); + + modelBuilder.Entity(entity => + { + entity.HasNoKey(); + entity.IsContinuousAggregate( + "trade_aggregate_filtered", + "1 hour", + x => x.Timestamp) + .AddAggregateFunction(x => x.AvgPrice, x => x.Price, EAggregateFunction.Avg) + .Where("\"Ticker\" = 'AAPL'"); + + // Map properties to view columns + entity.Property(x => x.TimeBucket).HasColumnName("time_bucket"); + entity.Property(x => x.AvgPrice).HasColumnName("AvgPrice"); + }); + } + } + + private class WithNoDataTestContext : DbContext + { + private readonly string _connectionString; + + public WithNoDataTestContext(string connectionString) + { + _connectionString = connectionString; + } + + public DbSet Trades => Set(); + public DbSet TradeAggregates => Set(); + + protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder) + { + optionsBuilder.UseNpgsql(_connectionString) + .UseTimescaleDb(); + } + + protected override void OnModelCreating(ModelBuilder modelBuilder) + { + modelBuilder.Entity(entity => + { + entity.ToTable("Trades"); + entity.HasNoKey(); + entity.IsHypertable(x => x.Timestamp); + }); + + modelBuilder.Entity(entity => + { + entity.HasNoKey(); + entity.IsContinuousAggregate( + "trade_aggregate_no_data", + "1 hour", + x => x.Timestamp) + .AddAggregateFunction(x => x.AvgPrice, x => x.Price, EAggregateFunction.Avg) + .WithNoData(true) + .MaterializedOnly(true); // Disable real-time aggregation so WITH NO DATA takes effect + + // Map properties to view columns + entity.Property(x => x.TimeBucket).HasColumnName("time_bucket"); + entity.Property(x => x.AvgPrice).HasColumnName("AvgPrice"); + }); + } + } + + private class CustomChunkIntervalTestContext : DbContext + { + private readonly string _connectionString; + + public CustomChunkIntervalTestContext(string connectionString) + { + _connectionString = connectionString; + } + + public DbSet Trades => Set(); + public DbSet TradeAggregates => Set(); + + protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder) + { + optionsBuilder.UseNpgsql(_connectionString) + .UseTimescaleDb(); + } + + protected override void OnModelCreating(ModelBuilder modelBuilder) + { + modelBuilder.Entity(entity => + { + entity.ToTable("Trades"); + entity.HasNoKey(); + entity.IsHypertable(x => x.Timestamp); + }); + + modelBuilder.Entity(entity => + { + entity.HasNoKey(); + entity.IsContinuousAggregate( + "trade_aggregate_custom_chunk", + "1 hour", + x => x.Timestamp, + chukInterval: "1 day") + .AddAggregateFunction(x => x.AvgPrice, x => x.Price, EAggregateFunction.Avg); + + // Map properties to view columns + entity.Property(x => x.TimeBucket).HasColumnName("time_bucket"); + entity.Property(x => x.AvgPrice).HasColumnName("AvgPrice"); + }); + } + } + + private class CreateGroupIndexesTestContext : DbContext + { + private readonly string _connectionString; + + public CreateGroupIndexesTestContext(string connectionString) + { + _connectionString = connectionString; + } + + public DbSet Trades => Set(); + public DbSet TradeAggregates => Set(); + + protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder) + { + optionsBuilder.UseNpgsql(_connectionString) + .UseTimescaleDb(); + } + + protected override void OnModelCreating(ModelBuilder modelBuilder) + { + modelBuilder.Entity(entity => + { + entity.ToTable("Trades"); + entity.HasNoKey(); + entity.IsHypertable(x => x.Timestamp); + }); + + modelBuilder.Entity(entity => + { + entity.HasNoKey(); + entity.IsContinuousAggregate( + "trade_aggregate_with_indexes", + "1 hour", + x => x.Timestamp) + .AddAggregateFunction(x => x.AvgPrice, x => x.Price, EAggregateFunction.Avg) + .AddGroupByColumn(x => x.Exchange) + .CreateGroupIndexes(true); + + // Map properties to view columns + entity.Property(x => x.TimeBucket).HasColumnName("time_bucket"); + entity.Property(x => x.Exchange).HasColumnName("Exchange"); + entity.Property(x => x.AvgPrice).HasColumnName("AvgPrice"); + }); + } + } + + private class MaterializedOnlyFalseTestContext : DbContext + { + private readonly string _connectionString; + + public MaterializedOnlyFalseTestContext(string connectionString) + { + _connectionString = connectionString; + } + + public DbSet Trades => Set(); + public DbSet TradeAggregates => Set(); + + protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder) + { + optionsBuilder.UseNpgsql(_connectionString) + .UseTimescaleDb(); + } + + protected override void OnModelCreating(ModelBuilder modelBuilder) + { + modelBuilder.Entity(entity => + { + entity.ToTable("Trades"); + entity.HasNoKey(); + entity.IsHypertable(x => x.Timestamp); + }); + + modelBuilder.Entity(entity => + { + entity.HasNoKey(); + entity.IsContinuousAggregate( + "trade_aggregate_realtime", + "1 hour", + x => x.Timestamp) + .AddAggregateFunction(x => x.AvgPrice, x => x.Price, EAggregateFunction.Avg) + .MaterializedOnly(false); + + // Map properties to view columns + entity.Property(x => x.TimeBucket).HasColumnName("time_bucket"); + entity.Property(x => x.AvgPrice).HasColumnName("AvgPrice"); + }); + } + } + + private class AlterChunkIntervalContext_Before : DbContext + { + private readonly string _connectionString; + + public AlterChunkIntervalContext_Before(string connectionString) + { + _connectionString = connectionString; + } + + public DbSet Trades => Set(); + public DbSet TradeAggregates => Set(); + + protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder) + { + optionsBuilder.UseNpgsql(_connectionString) + .UseTimescaleDb(); + } + + protected override void OnModelCreating(ModelBuilder modelBuilder) + { + modelBuilder.Entity(entity => + { + entity.ToTable("Trades"); + entity.HasNoKey(); + entity.IsHypertable(x => x.Timestamp); + }); + + modelBuilder.Entity(entity => + { + entity.HasNoKey(); + entity.IsContinuousAggregate( + "trade_aggregate_alterable", + "1 hour", + x => x.Timestamp, + chukInterval: "7 days") + .AddAggregateFunction(x => x.AvgPrice, x => x.Price, EAggregateFunction.Avg); + + // Map properties to view columns + entity.Property(x => x.TimeBucket).HasColumnName("time_bucket"); + entity.Property(x => x.AvgPrice).HasColumnName("AvgPrice"); + }); + } + } + + private class AlterChunkIntervalContext_After : DbContext + { + private readonly string _connectionString; + + public AlterChunkIntervalContext_After(string connectionString) + { + _connectionString = connectionString; + } + + public DbSet Trades => Set(); + public DbSet TradeAggregates => Set(); + + protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder) + { + optionsBuilder.UseNpgsql(_connectionString) + .UseTimescaleDb(); + } + + protected override void OnModelCreating(ModelBuilder modelBuilder) + { + modelBuilder.Entity(entity => + { + entity.ToTable("Trades"); + entity.HasNoKey(); + entity.IsHypertable(x => x.Timestamp); + }); + + modelBuilder.Entity(entity => + { + entity.HasNoKey(); + entity.IsContinuousAggregate( + "trade_aggregate_alterable", + "1 hour", + x => x.Timestamp, + chukInterval: "14 days") + .AddAggregateFunction(x => x.AvgPrice, x => x.Price, EAggregateFunction.Avg); + + // Map properties to view columns + entity.Property(x => x.TimeBucket).HasColumnName("time_bucket"); + entity.Property(x => x.AvgPrice).HasColumnName("AvgPrice"); + }); + } + } + + private class AlterMaterializedOnlyTestContext : DbContext + { + private readonly string _connectionString; + + public AlterMaterializedOnlyTestContext(string connectionString) + { + _connectionString = connectionString; + } + + public DbSet Trades => Set(); + public DbSet TradeAggregates => Set(); + + protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder) + { + optionsBuilder.UseNpgsql(_connectionString) + .UseTimescaleDb(); + } + + protected override void OnModelCreating(ModelBuilder modelBuilder) + { + modelBuilder.Entity(entity => + { + entity.ToTable("Trades"); + entity.HasNoKey(); + entity.IsHypertable(x => x.Timestamp); + }); + + modelBuilder.Entity(entity => + { + entity.HasNoKey(); + entity.IsContinuousAggregate( + "trade_aggregate_materialized_only", + "1 hour", + x => x.Timestamp) + .AddAggregateFunction(x => x.AvgPrice, x => x.Price, EAggregateFunction.Avg) + .MaterializedOnly(false); + + // Map properties to view columns + entity.Property(x => x.TimeBucket).HasColumnName("time_bucket"); + entity.Property(x => x.AvgPrice).HasColumnName("AvgPrice"); + }); + } + } + + private class AlterCreateGroupIndexesTestContext : DbContext + { + private readonly string _connectionString; + + public AlterCreateGroupIndexesTestContext(string connectionString) + { + _connectionString = connectionString; + } + + public DbSet Trades => Set(); + public DbSet TradeAggregates => Set(); + + protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder) + { + optionsBuilder.UseNpgsql(_connectionString) + .UseTimescaleDb(); + } + + protected override void OnModelCreating(ModelBuilder modelBuilder) + { + modelBuilder.Entity(entity => + { + entity.ToTable("Trades"); + entity.HasNoKey(); + entity.IsHypertable(x => x.Timestamp); + }); + + modelBuilder.Entity(entity => + { + entity.HasNoKey(); + entity.IsContinuousAggregate( + "trade_aggregate_group_indexes", + "1 hour", + x => x.Timestamp) + .AddAggregateFunction(x => x.AvgPrice, x => x.Price, EAggregateFunction.Avg) + .AddGroupByColumn(x => x.Exchange) + .CreateGroupIndexes(false); + + // Map properties to view columns + entity.Property(x => x.TimeBucket).HasColumnName("time_bucket"); + entity.Property(x => x.Exchange).HasColumnName("Exchange"); + entity.Property(x => x.AvgPrice).HasColumnName("AvgPrice"); + }); + } + } + + private class DropTestContext : DbContext + { + private readonly string _connectionString; + + public DropTestContext(string connectionString) + { + _connectionString = connectionString; + } + + public DbSet Trades => Set(); + public DbSet TradeAggregates => Set(); + + protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder) + { + optionsBuilder.UseNpgsql(_connectionString) + .UseTimescaleDb(); + } + + protected override void OnModelCreating(ModelBuilder modelBuilder) + { + modelBuilder.Entity(entity => + { + entity.ToTable("Trades"); + entity.HasNoKey(); + entity.IsHypertable(x => x.Timestamp); + }); + + modelBuilder.Entity(entity => + { + entity.HasNoKey(); + entity.IsContinuousAggregate( + "trade_aggregate_to_drop", + "1 hour", + x => x.Timestamp) + .AddAggregateFunction(x => x.AvgPrice, x => x.Price, EAggregateFunction.Avg); + + // Map properties to view columns + entity.Property(x => x.TimeBucket).HasColumnName("time_bucket"); + entity.Property(x => x.AvgPrice).HasColumnName("AvgPrice"); + }); + } + } + + private class SqlGenerationTestContext : DbContext + { + private readonly string _connectionString; + + public SqlGenerationTestContext(string connectionString) + { + _connectionString = connectionString; + } + + public DbSet Trades => Set(); + public DbSet TradeAggregates => Set(); + + protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder) + { + optionsBuilder.UseNpgsql(_connectionString) + .UseTimescaleDb(); + } + + protected override void OnModelCreating(ModelBuilder modelBuilder) + { + modelBuilder.Entity(entity => + { + entity.ToTable("Trades"); + entity.HasNoKey(); + entity.IsHypertable(x => x.Timestamp); + }); + + modelBuilder.Entity(entity => + { + entity.HasNoKey(); + entity.IsContinuousAggregate( + "trade_aggregate_sql_gen", + "1 hour", + x => x.Timestamp) + .AddAggregateFunction(x => x.AvgPrice, x => x.Price, EAggregateFunction.Avg); + + // Map properties to view columns + entity.Property(x => x.TimeBucket).HasColumnName("time_bucket"); + entity.Property(x => x.AvgPrice).HasColumnName("AvgPrice"); + }); + } + } + + private class SnakeCaseTestContext : DbContext + { + private readonly string _connectionString; + + public SnakeCaseTestContext(string connectionString) + { + _connectionString = connectionString; + } + + public DbSet Trades => Set(); + public DbSet TradeAggregates => Set(); + + protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder) + { + optionsBuilder.UseNpgsql(_connectionString) + .UseSnakeCaseNamingConvention() + .UseTimescaleDb(); + } + + protected override void OnModelCreating(ModelBuilder modelBuilder) + { + modelBuilder.Entity(entity => + { + entity.ToTable("trades"); + entity.HasNoKey(); + entity.IsHypertable(x => x.Timestamp); + }); + + modelBuilder.Entity(entity => + { + entity.HasNoKey(); + entity.IsContinuousAggregate( + "trade_aggregate_snake_case", + "1 hour", + x => x.Timestamp) + .AddAggregateFunction(x => x.avg_price, x => x.Price, EAggregateFunction.Avg); + + // Note: snake_case convention is applied automatically, so time_bucket and avg_price are already correct + }); + } + } + + #endregion + } +} diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB.sln b/CmdScale.EntityFrameworkCore.TimescaleDB.sln index 46427ad..46d72e0 100644 --- a/CmdScale.EntityFrameworkCore.TimescaleDB.sln +++ b/CmdScale.EntityFrameworkCore.TimescaleDB.sln @@ -13,6 +13,7 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "CmdScale.EntityFrameworkCor EndProject Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution Items", "{8EC462FD-D22E-90A8-E5CE-7E832BA40C5D}" ProjectSection(SolutionItems) = preProject + .editorconfig = .editorconfig .gitignore = .gitignore CODE_OF_CONDUCT.md = CODE_OF_CONDUCT.md docker-compose.yml = docker-compose.yml diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB/Abstractions/EAggregateFunction.cs b/CmdScale.EntityFrameworkCore.TimescaleDB/Abstractions/EAggregateFunction.cs new file mode 100644 index 0000000..3e3c5c7 --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB/Abstractions/EAggregateFunction.cs @@ -0,0 +1,13 @@ +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Abstractions +{ + public enum EAggregateFunction + { + Avg, + Sum, + Min, + Max, + Count, + First, + Last + } +} diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB/Configuration/ContinuousAggregate/AggregateAttribute.cs b/CmdScale.EntityFrameworkCore.TimescaleDB/Configuration/ContinuousAggregate/AggregateAttribute.cs new file mode 100644 index 0000000..43a45ae --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB/Configuration/ContinuousAggregate/AggregateAttribute.cs @@ -0,0 +1,22 @@ +using CmdScale.EntityFrameworkCore.TimescaleDB.Abstractions; + +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Configuration.ContinuousAggregate +{ + /// + /// Defines an aggregate function for a property. + /// + [AttributeUsage(AttributeTargets.Property)] + public class AggregateAttribute(EAggregateFunction function, string sourceColumn = "*") : Attribute + { + /// + /// The aggregate function to apply. + /// + public EAggregateFunction Function { get; } = function; + + /// + /// The name of the column in the source hypertable to aggregate. + /// For COUNT(*), this can be null or "*". + /// + public string SourceColumn { get; } = sourceColumn; + } +} diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB/Configuration/ContinuousAggregate/ContinuousAggregateAnnotations.cs b/CmdScale.EntityFrameworkCore.TimescaleDB/Configuration/ContinuousAggregate/ContinuousAggregateAnnotations.cs new file mode 100644 index 0000000..824c472 --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB/Configuration/ContinuousAggregate/ContinuousAggregateAnnotations.cs @@ -0,0 +1,24 @@ +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Configuration.ContinuousAggregate +{ + /// + /// Contains constants for annotations used by the TimescaleDB provider extension. + /// + public static class ContinuousAggregateAnnotations + { + public const string MaterializedViewName = "TimescaleDB:MaterializedViewName"; + public const string ParentName = "TimescaleDB:ParentName"; + public const string ChunkInterval = "TimescaleDB:ChunkInterval"; + + public const string WithNoData = "TimescaleDB:WithNoData"; + public const string CreateGroupIndexes = "TimescaleDB:CreateGroupIndexes"; + public const string MaterializedOnly = "TimescaleDB:MaterializedOnly"; + + public const string TimeBucketWidth = "TimescaleDB:TimeBucket:BucketWidth"; + public const string TimeBucketSourceColumn = "TimescaleDB:TimeBucket:SourceColumn"; + public const string TimeBucketGroupBy = "TimescaleDB:TimeBucket:GroupBy"; + + public const string AggregateFunctions = "TimescaleDB:AggregateFunctions"; + public const string WhereClause = "TimescaleDB:WhereClause"; + public const string GroupByColumns = "TimescaleDB:GroupByColumns"; + } +} diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB/Configuration/ContinuousAggregate/ContinuousAggregateAttribute.cs b/CmdScale.EntityFrameworkCore.TimescaleDB/Configuration/ContinuousAggregate/ContinuousAggregateAttribute.cs new file mode 100644 index 0000000..17c1012 --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB/Configuration/ContinuousAggregate/ContinuousAggregateAttribute.cs @@ -0,0 +1,68 @@ +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Configuration.ContinuousAggregate +{ + /// + /// Defines a TimescaleDB continuous aggregate on an EF Core entity. + /// This attribute provides all the necessary metadata to construct the + /// CREATE MATERIALIZED VIEW statement for a continuous aggregate. + /// + [AttributeUsage(AttributeTargets.Class)] + public class ContinuousAggregateAttribute : Attribute + { + /// + /// Gets or sets the name of the materialized view that will be created in the database. + /// This corresponds to the in the CREATE MATERIALIZED VIEW statement. + /// + public string MaterializedViewName { get; set; } = string.Empty; + + /// + /// Gets or sets the name of the source hypertable or another continuous aggregate + /// on which this continuous aggregate is based. + /// + public string ParentName { get; set; } = string.Empty; + + /// + /// Gets or sets the chunk interval for the continuous aggregate's underlying materialized hypertable. + /// If not set, it defaults to 10 times the chunk_time_interval of the parent hypertable. + /// Corresponds to the 'timescaledb.chunk_interval' option. + /// + public string? ChunkInterval { get; set; } + + /// + /// By default, when you create a view for the first time, it is populated with data. This is so that the aggregates can be computed across the entire hypertable. + /// If you don't want this to happen, for example if the table is very large, or if new data is being continuously added, you can control the order in which the data is refreshed. + /// You can do this by adding a manual refresh with your continuous aggregate policy using the WITH NO DATA option. + /// + public bool WithNoData { get; set; } = false; + + /// + /// Gets or sets a value indicating whether to automatically create indexes on the GROUP BY columns. + /// Defaults to true. Corresponds to the 'timescaledb.create_group_indexes' option. + /// + public bool CreateGroupIndexes { get; set; } = true; + + /// + /// Gets or sets a value indicating whether queries to the view should only return materialized data. + /// If false (the default), recent data from the source hypertable that has not yet been materialized will be included in query results. + /// Corresponds to the 'timescaledb.materialized_only' option. + /// + public bool MaterializedOnly { get; set; } = false; + + /// + /// Gets or sets the time interval for the time_bucket function (e.g., "1 day", "15 minutes"). + /// This is a required parameter for defining the aggregation window. + /// + public string TimeBucketWidth { get; set; } = string.Empty; + + /// + /// Gets or sets the name of the time column in the source hypertable to be used by the time_bucket function. + /// + public string TimeBucketSourceColumn { get; set; } = string.Empty; + + + /// + /// Gets or sets an optional SQL WHERE clause to filter rows from the source hypertable before aggregation. + /// The clause should be a valid SQL string without the "WHERE" keyword itself (e.g., "device_id = 'sensor-1'"). + /// + public string? Where { get; set; } + } +} diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB/Configuration/ContinuousAggregate/ContinuousAggregateBuilder.cs b/CmdScale.EntityFrameworkCore.TimescaleDB/Configuration/ContinuousAggregate/ContinuousAggregateBuilder.cs new file mode 100644 index 0000000..6cd27f3 --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB/Configuration/ContinuousAggregate/ContinuousAggregateBuilder.cs @@ -0,0 +1,22 @@ +using Microsoft.EntityFrameworkCore.Metadata.Builders; + +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Configuration.ContinuousAggregate +{ + /// + /// Provides a fluent API for configuring a TimescaleDB continuous aggregate. + /// This builder is aware of both the aggregate entity type and the source hypertable entity type. + /// + /// The class representing the continuous aggregate view. + /// The class representing the source hypertable. + public class ContinuousAggregateBuilder + where TEntity : class + where TSourceEntity : class + { + public EntityTypeBuilder EntityTypeBuilder { get; } + + internal ContinuousAggregateBuilder(EntityTypeBuilder entityTypeBuilder) + { + EntityTypeBuilder = entityTypeBuilder; + } + } +} diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB/Configuration/ContinuousAggregate/ContinuousAggregateConvention.cs b/CmdScale.EntityFrameworkCore.TimescaleDB/Configuration/ContinuousAggregate/ContinuousAggregateConvention.cs new file mode 100644 index 0000000..0740d2e --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB/Configuration/ContinuousAggregate/ContinuousAggregateConvention.cs @@ -0,0 +1,71 @@ +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Metadata; +using Microsoft.EntityFrameworkCore.Metadata.Builders; +using Microsoft.EntityFrameworkCore.Metadata.Conventions; +using System.Reflection; + +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Configuration.ContinuousAggregate +{ + /// + /// Reads the [ContinuousAggregate], [TimeBucket], and [Aggregate] attributes + /// to configure an entity as a TimescaleDB continuous aggregate. + /// + public class ContinuousAggregateConvention : IEntityTypeAddedConvention + { + public void ProcessEntityTypeAdded(IConventionEntityTypeBuilder entityTypeBuilder, IConventionContext context) + { + IConventionEntityType entityType = entityTypeBuilder.Metadata; + ContinuousAggregateAttribute? continuousAggregateAttribute = entityType.ClrType?.GetCustomAttribute(); + + if (continuousAggregateAttribute == null) return; + + // Configure the entity to map to a view instead of a table + // This prevents EF Core from trying to create a table for the continuous aggregate + entityTypeBuilder.ToView(continuousAggregateAttribute.MaterializedViewName); + + // Apply class-level configurations from [ContinuousAggregateAttribute] + entityTypeBuilder.HasAnnotation(ContinuousAggregateAnnotations.MaterializedViewName, continuousAggregateAttribute.MaterializedViewName); + entityTypeBuilder.HasAnnotation(ContinuousAggregateAnnotations.ParentName, continuousAggregateAttribute.ParentName); + entityTypeBuilder.HasAnnotation(ContinuousAggregateAnnotations.ChunkInterval, continuousAggregateAttribute.ChunkInterval); + entityTypeBuilder.HasAnnotation(ContinuousAggregateAnnotations.WithNoData, continuousAggregateAttribute.WithNoData); + entityTypeBuilder.HasAnnotation(ContinuousAggregateAnnotations.CreateGroupIndexes, continuousAggregateAttribute.CreateGroupIndexes); + entityTypeBuilder.HasAnnotation(ContinuousAggregateAnnotations.MaterializedOnly, continuousAggregateAttribute.MaterializedOnly); + entityTypeBuilder.HasAnnotation(ContinuousAggregateAnnotations.WhereClause, continuousAggregateAttribute.Where); + + // Discover class-level TimeBucket configuration from [TimeBucketAttribute] + TimeBucketAttribute? timeBucketAttr = entityType.ClrType?.GetCustomAttribute(); + if (timeBucketAttr != null) + { + entityTypeBuilder.HasAnnotation(ContinuousAggregateAnnotations.TimeBucketWidth, timeBucketAttr.BucketWidth); + entityTypeBuilder.HasAnnotation(ContinuousAggregateAnnotations.TimeBucketSourceColumn, timeBucketAttr.SourceColumn); + entityTypeBuilder.HasAnnotation(ContinuousAggregateAnnotations.TimeBucketGroupBy, timeBucketAttr.GroupBy); + } + + // Discover property-level configurations + List aggregateFunctions = []; + + foreach (IConventionProperty property in entityType.GetProperties()) + { + PropertyInfo? propertyInfo = property.PropertyInfo; + if (propertyInfo == null) continue; + + // Discover aggregate columns from [AggregateAttribute] + AggregateAttribute? aggregateAttr = propertyInfo.GetCustomAttribute(); + if (aggregateAttr != null) + { + // Serialize the aggregate info into a string format for the annotation. + // Format: "DestinationPropertyName:AggregateFunction:SourceColumnName" + // Example: "AvgTemperature:Avg:temperature" + string sourceColumn = aggregateAttr.SourceColumn ?? property.Name; + aggregateFunctions.Add($"{property.Name}:{aggregateAttr.Function}:{sourceColumn}"); + } + } + + // Apply the discovered property-level annotations + if (aggregateFunctions.Count != 0) + { + entityTypeBuilder.HasAnnotation(ContinuousAggregateAnnotations.AggregateFunctions, aggregateFunctions); + } + } + } +} diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB/Configuration/ContinuousAggregate/ContinuousAggregateTypeBuilder.cs b/CmdScale.EntityFrameworkCore.TimescaleDB/Configuration/ContinuousAggregate/ContinuousAggregateTypeBuilder.cs new file mode 100644 index 0000000..5a0d09b --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB/Configuration/ContinuousAggregate/ContinuousAggregateTypeBuilder.cs @@ -0,0 +1,176 @@ +using CmdScale.EntityFrameworkCore.TimescaleDB.Abstractions; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; +using Microsoft.EntityFrameworkCore.Metadata.Builders; +using System.Linq.Expressions; + +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Configuration.ContinuousAggregate +{ + public static class ContinuousAggregateTypeBuilder + { + public static ContinuousAggregateBuilder IsContinuousAggregate( + this EntityTypeBuilder entityTypeBuilder, + string materualizedViewName, + string timeBucketWidth, + Expression> propertyExpression, + bool timeBucketGroupBy = true, + string? chukInterval = null) + where TEntity : class + where TSourceEntity : class + { + // Configure the entity to map to a view instead of a table + // This prevents EF Core from trying to create a table for the continuous aggregate + entityTypeBuilder.ToView(materualizedViewName); + + entityTypeBuilder.HasAnnotation(ContinuousAggregateAnnotations.MaterializedViewName, materualizedViewName); + + string parentName = typeof(TSourceEntity).Name; + entityTypeBuilder.HasAnnotation(ContinuousAggregateAnnotations.ParentName, parentName); + + string timeBucketSourceColumn = GetPropertyName(propertyExpression); + entityTypeBuilder.HasAnnotation(ContinuousAggregateAnnotations.TimeBucketSourceColumn, timeBucketSourceColumn); + entityTypeBuilder.HasAnnotation(ContinuousAggregateAnnotations.TimeBucketWidth, timeBucketWidth); + entityTypeBuilder.HasAnnotation(ContinuousAggregateAnnotations.TimeBucketGroupBy, timeBucketGroupBy); + + if (!string.IsNullOrEmpty(chukInterval)) + { + entityTypeBuilder.HasAnnotation(ContinuousAggregateAnnotations.ChunkInterval, chukInterval); + } + + return new ContinuousAggregateBuilder(entityTypeBuilder); + } + + public static ContinuousAggregateBuilder WithNoData( + this ContinuousAggregateBuilder builder, + bool withNoData = true) + where TEntity : class + where TSourceEntity : class + { + + builder.EntityTypeBuilder.HasAnnotation(ContinuousAggregateAnnotations.WithNoData, withNoData); + return builder; + } + + public static ContinuousAggregateBuilder CreateGroupIndexes( + this ContinuousAggregateBuilder builder, + bool createGroupIndexes = true) + where TEntity : class + where TSourceEntity : class + { + builder.EntityTypeBuilder.HasAnnotation(ContinuousAggregateAnnotations.CreateGroupIndexes, createGroupIndexes); + return builder; + } + + public static ContinuousAggregateBuilder MaterializedOnly( + this ContinuousAggregateBuilder builder, + bool materializedOnly = true) + where TEntity : class + where TSourceEntity : class + { + builder.EntityTypeBuilder.HasAnnotation(ContinuousAggregateAnnotations.MaterializedOnly, materializedOnly); + return builder; + } + + public static ContinuousAggregateBuilder AddAggregateFunction( + this ContinuousAggregateBuilder builder, + Expression> propertyExpression, + Expression> sourceColumn, + EAggregateFunction function + ) + where TEntity : class + where TSourceEntity : class + { + string propertyName = GetPropertyName(propertyExpression); + IAnnotation? annotation = builder.EntityTypeBuilder.Metadata.FindAnnotation(ContinuousAggregateAnnotations.AggregateFunctions); + List aggregateFunctions = annotation?.Value as List ?? []; + + if (aggregateFunctions.Any(x => x.StartsWith(propertyName + ":"))) + { + return builder; + } + + string sourceColumnName = GetPropertyName(sourceColumn); + + aggregateFunctions.Add($"{propertyName}:{function}:{sourceColumnName}"); + builder.EntityTypeBuilder.HasAnnotation(ContinuousAggregateAnnotations.AggregateFunctions, aggregateFunctions); + return builder; + } + + public static ContinuousAggregateBuilder AddGroupByColumn( + this ContinuousAggregateBuilder builder, + Expression> propertyExpression) + where TEntity : class + where TSourceEntity : class + { + string propertyName = GetPropertyName(propertyExpression); + IAnnotation? annotation = builder.EntityTypeBuilder.Metadata.FindAnnotation(ContinuousAggregateAnnotations.GroupByColumns); + List groupByColumns = annotation?.Value as List ?? []; + + if (groupByColumns.Contains(propertyName)) + { + return builder; + } + + groupByColumns.Add(propertyName); + + builder.EntityTypeBuilder.HasAnnotation(ContinuousAggregateAnnotations.GroupByColumns, groupByColumns); + return builder; + } + + public static ContinuousAggregateBuilder AddGroupByColumn( + this ContinuousAggregateBuilder builder, + string groupByExpression) + where TEntity : class + where TSourceEntity : class + { + IAnnotation? annotation = builder.EntityTypeBuilder.Metadata.FindAnnotation(ContinuousAggregateAnnotations.GroupByColumns); + List groupByColumns = annotation?.Value as List ?? []; + + if (groupByColumns.Contains(groupByExpression)) + { + return builder; + } + + groupByColumns.Add(groupByExpression); + + builder.EntityTypeBuilder.HasAnnotation(ContinuousAggregateAnnotations.GroupByColumns, groupByColumns); + return builder; + } + + // TODO: Remove or implement expression parsing + //public static ContinuousAggregateBuilder Where( + // this ContinuousAggregateBuilder builder, + // Expression> predicate) + // where TEntity : class + // where TSourceEntity : class + //{ + // builder.EntityTypeBuilder.HasAnnotation(ContinuousAggregateAnnotations.WhereClause, predicate); + // return builder; + //} + + public static ContinuousAggregateBuilder Where( + this ContinuousAggregateBuilder builder, + string whereClause) + where TEntity : class + where TSourceEntity : class + { + builder.EntityTypeBuilder.HasAnnotation(ContinuousAggregateAnnotations.WhereClause, whereClause); + return builder; + } + + private static string GetPropertyName(Expression> propertyExpression) + { + if (propertyExpression.Body is MemberExpression memberExpression) + { + return memberExpression.Member.Name; + } + + if (propertyExpression.Body is UnaryExpression unaryExpression && unaryExpression.Operand is MemberExpression unaryMemberExpression) + { + return unaryMemberExpression.Member.Name; + } + + throw new ArgumentException("Expression must be a simple property access expression.", nameof(propertyExpression)); + } + } +} diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB/Configuration/ContinuousAggregate/TimeBucketAttribute.cs b/CmdScale.EntityFrameworkCore.TimescaleDB/Configuration/ContinuousAggregate/TimeBucketAttribute.cs new file mode 100644 index 0000000..24b5e4b --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB/Configuration/ContinuousAggregate/TimeBucketAttribute.cs @@ -0,0 +1,29 @@ +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Configuration.ContinuousAggregate +{ + /// + /// Define the time bucket column for a continuous aggregate. + /// + /// + /// Initializes a new instance of the class. + /// + /// The time interval for the bucket (e.g., "1 hour", "15 minutes"). + /// The name of the time column in the source hypertable. + [AttributeUsage(AttributeTargets.Class)] + public class TimeBucketAttribute(string bucketWidth, string sourceColumn) : Attribute + { + /// + /// The time interval for the bucket (e.g., "1 hour", "15 minutes"). + /// + public string BucketWidth { get; } = bucketWidth; + + /// + /// The name of the time column in the source hypertable. + /// + public string SourceColumn { get; } = sourceColumn; + + /// + /// Weither the time bucket column should be included in the GROUP BY clause. + /// + public bool GroupBy { get; set; } = true; + } +} diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB/Configuration/ReorderPolicy/ReorderPolicyTypeBulder.cs b/CmdScale.EntityFrameworkCore.TimescaleDB/Configuration/ReorderPolicy/ReorderPolicyTypeBuilder.cs similarity index 100% rename from CmdScale.EntityFrameworkCore.TimescaleDB/Configuration/ReorderPolicy/ReorderPolicyTypeBulder.cs rename to CmdScale.EntityFrameworkCore.TimescaleDB/Configuration/ReorderPolicy/ReorderPolicyTypeBuilder.cs diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB/Generators/ContinuousAggregateOperationGenerator.cs b/CmdScale.EntityFrameworkCore.TimescaleDB/Generators/ContinuousAggregateOperationGenerator.cs new file mode 100644 index 0000000..833419c --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB/Generators/ContinuousAggregateOperationGenerator.cs @@ -0,0 +1,232 @@ +using CmdScale.EntityFrameworkCore.TimescaleDB.Operations; + +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Generators +{ + public class ContinuousAggregateOperationGenerator + { + private readonly string quoteString = "\""; + private readonly SqlBuilderHelper sqlHelper; + + public ContinuousAggregateOperationGenerator(bool isDesignTime = false) + { + if (isDesignTime) + { + quoteString = "\"\""; + } + + sqlHelper = new SqlBuilderHelper(quoteString); + + } + + public List Generate(CreateContinuousAggregateOperation operation) + { + string qualifiedIdentifier = sqlHelper.QualifiedIdentifier(operation.MaterializedViewName, operation.Schema); + string parentQualifiedIdentifier = sqlHelper.QualifiedIdentifier(operation.ParentName, operation.Schema); + + List statements = []; + + // Build WITH options + List withOptions = + [ + "timescaledb.continuous", + $"timescaledb.create_group_indexes = {operation.CreateGroupIndexes.ToString().ToLower()}", + $"timescaledb.materialized_only = {operation.MaterializedOnly.ToString().ToLower()}" + ]; + + // Add optional chunk_interval if specified + if (!string.IsNullOrEmpty(operation.ChunkInterval)) + { + withOptions.Add($"timescaledb.chunk_interval = '{operation.ChunkInterval}'"); + } + + // Build the SELECT list + List selectList = []; + + // Add time_bucket column + string timeBucketColumn = $"{quoteString}{operation.TimeBucketSourceColumn}{quoteString}"; + string timeBucketWidthSql = $"'{operation.TimeBucketWidth}'"; + selectList.Add($"time_bucket({timeBucketWidthSql}, {timeBucketColumn}) AS time_bucket"); + + // Add GROUP BY columns to SELECT (only actual columns, not SQL expressions) + foreach (string groupByColumn in operation.GroupByColumns) + { + // Check if it's a raw SQL expression or a column name + bool isRawSqlExpression = groupByColumn.Contains(',') || groupByColumn.Contains('(') || groupByColumn.Contains(' '); + if (!isRawSqlExpression) + { + selectList.Add($"{quoteString}{groupByColumn}{quoteString}"); + } + } + + // Build aggregate functions + foreach (string aggInfo in operation.AggregateFunctions) + { + string[] parts = aggInfo.Split(':'); + if (parts.Length != 3) + { + // Skip malformed string + continue; + } + + string alias = parts[0]; + string functionEnumString = parts[1]; + string sourceColumn = parts[2]; + + string sqlFunction = GetSqlAggregateFunction(functionEnumString); + string quotedSourceColumn = $"{quoteString}{sourceColumn}{quoteString}"; + string quotedAlias = $"{quoteString}{alias}{quoteString}"; + string aggregateExpression; + + // Handle special TimescaleDB aggregates 'first' and 'last' + // which require (value_column, time_column) + if (sqlFunction == "first" || sqlFunction == "last") + { + aggregateExpression = $"{sqlFunction}({quotedSourceColumn}, {timeBucketColumn})"; + } + else + { + aggregateExpression = $"{sqlFunction}({quotedSourceColumn})"; + } + + selectList.Add($"{aggregateExpression} AS {quotedAlias}"); + } + + // Build the GROUP BY list + List groupByList = []; + if (operation.TimeBucketGroupBy) + { + groupByList.Add("time_bucket"); + } + + // Add group by columns + foreach (string groupByColumn in operation.GroupByColumns) + { + if (groupByColumn.Contains(',') || groupByColumn.Contains('(') || groupByColumn.Contains(' ')) + { + // It's a raw SQL expression, use as-is + groupByList.Add(groupByColumn); + } + else + { + // It's a column name, quote it + groupByList.Add($"{quoteString}{groupByColumn}{quoteString}"); + } + } + + // Build the complete CREATE MATERIALIZED VIEW statement as a single string + var sqlBuilder = new System.Text.StringBuilder(); + sqlBuilder.Append($"CREATE MATERIALIZED VIEW {qualifiedIdentifier}"); + sqlBuilder.AppendLine(); + sqlBuilder.Append($"WITH ({string.Join(", ", withOptions)}) AS"); + sqlBuilder.AppendLine(); + sqlBuilder.Append($"SELECT {string.Join(", ", selectList)}"); + sqlBuilder.AppendLine(); + sqlBuilder.Append($"FROM {parentQualifiedIdentifier}"); + + // Add WHERE clause if specified + if (!string.IsNullOrWhiteSpace(operation.WhereClaus)) + { + string whereClause = operation.WhereClaus.Replace("\"", quoteString); + sqlBuilder.AppendLine(); + sqlBuilder.Append($"WHERE {whereClause}"); + } + + // Add GROUP BY clause + if (groupByList.Count > 0) + { + sqlBuilder.AppendLine(); + sqlBuilder.Append($"GROUP BY {string.Join(", ", groupByList)}"); + } + + // Add WITH [NO] DATA + if (operation.WithNoData) + { + sqlBuilder.AppendLine(); + sqlBuilder.Append("WITH NO DATA"); + } + + sqlBuilder.Append(';'); + statements.Add(sqlBuilder.ToString()); + + return statements; + } + + public List Generate(AlterContinuousAggregateOperation operation) + { + string qualifiedIdentifier = sqlHelper.QualifiedIdentifier(operation.MaterializedViewName, operation.Schema); + List statements = []; + + // Check for ChunkInterval change + // Note: TimescaleDB continuous aggregates only support SET for chunk_interval, not RESET + if (operation.ChunkInterval != operation.OldChunkInterval) + { + // Only generate SQL if we have a valid new value to set + // We cannot RESET chunk_interval as TimescaleDB doesn't support it + if (!string.IsNullOrEmpty(operation.ChunkInterval)) + { + string chunkIntervalSql = $"'{operation.ChunkInterval}'"; + statements.Add($"ALTER MATERIALIZED VIEW {qualifiedIdentifier} SET (timescaledb.chunk_interval = {chunkIntervalSql});"); + } + else if (!string.IsNullOrEmpty(operation.OldChunkInterval)) + { + // Special case: If new value is null/empty but old value exists, + // restore the old value instead of trying to RESET (which is unsupported) + string chunkIntervalSql = $"'{operation.OldChunkInterval}'"; + statements.Add($"ALTER MATERIALIZED VIEW {qualifiedIdentifier} SET (timescaledb.chunk_interval = {chunkIntervalSql});"); + } + } + + // Check for CreateGroupIndexes change + if (operation.CreateGroupIndexes != operation.OldCreateGroupIndexes) + { + string createGroupIndexesValue = operation.CreateGroupIndexes.ToString().ToLower(); + statements.Add($"ALTER MATERIALIZED VIEW {qualifiedIdentifier} SET (timescaledb.create_group_indexes = {createGroupIndexesValue});"); + } + + // Check for MaterializedOnly change + if (operation.MaterializedOnly != operation.OldMaterializedOnly) + { + string materializedOnlyValue = operation.MaterializedOnly.ToString().ToLower(); + statements.Add($"ALTER MATERIALIZED VIEW {qualifiedIdentifier} SET (timescaledb.materialized_only = {materializedOnlyValue});"); + } + + return statements; + } + + public List Generate(DropContinuousAggregateOperation operation) + { + string qualifiedIdentifier = sqlHelper.QualifiedIdentifier(operation.MaterializedViewName, operation.Schema); + List statements = []; + + statements.Add($"DROP MATERIALIZED VIEW IF EXISTS {qualifiedIdentifier};"); + + return statements; + } + + /// + /// Translates the string representation of EAggregateFunction into a SQL function. + /// + private static string GetSqlAggregateFunction(string functionEnumString) + { + switch (functionEnumString) + { + case "Avg": + return "AVG"; + case "Max": + return "MAX"; + case "Min": + return "MIN"; + case "Sum": + return "SUM"; + case "Count": + return "COUNT"; + case "First": + return "first"; + case "Last": + return "last"; + default: + throw new NotSupportedException($"The aggregate function '{functionEnumString}' is not supported by the generator."); + } + } + } +} diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB/Generators/SqlBuilderHelper.cs b/CmdScale.EntityFrameworkCore.TimescaleDB/Generators/SqlBuilderHelper.cs index 8c25240..aab9cf3 100644 --- a/CmdScale.EntityFrameworkCore.TimescaleDB/Generators/SqlBuilderHelper.cs +++ b/CmdScale.EntityFrameworkCore.TimescaleDB/Generators/SqlBuilderHelper.cs @@ -7,17 +7,46 @@ public class SqlBuilderHelper(string quoteString) { private readonly string quoteString = quoteString; - public static void BuildQueryString(List statements, MigrationCommandListBuilder builder) + public static void BuildQueryString(List statements, MigrationCommandListBuilder builder, bool suppressTransaction = false) { + if (statements.Count == 0) + { + return; + } + + // Group consecutive statements that don't end with semicolon into single commands + List> commandGroups = []; + List currentGroup = []; + foreach (string statement in statements) { + currentGroup.Add(statement); + + // If statement ends with semicolon, it's a complete command + if (statement.TrimEnd().EndsWith(';')) + { + commandGroups.Add([.. currentGroup]); + currentGroup.Clear(); + } + } + + // Add any remaining statements as a final command + if (currentGroup.Count > 0) + { + commandGroups.Add([.. currentGroup]); + } + + // Build each command group + foreach (List group in commandGroups) + { + string command = string.Join("\n", group); builder - .Append(statement) - .EndCommand(); + .Append(command) + .EndCommand(suppressTransaction: suppressTransaction); } } - public static void BuildQueryString(List statements, IndentedStringBuilder builder) + public static void BuildQueryString(List statements, IndentedStringBuilder builder, bool suppressTransaction = false) { if (statements.Count > 0) { @@ -29,7 +58,14 @@ public static void BuildQueryString(List statements, IndentedStringBuild builder.AppendLine(statement); } } - builder.Append("\")"); + if (suppressTransaction) + { + builder.Append("\", suppressTransaction: true)"); + } + else + { + builder.Append("\")"); + } } } diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB/Internals/Features/ContinuousAggregates/ContinuousAggregateDiffer.cs b/CmdScale.EntityFrameworkCore.TimescaleDB/Internals/Features/ContinuousAggregates/ContinuousAggregateDiffer.cs new file mode 100644 index 0000000..3a11c51 --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB/Internals/Features/ContinuousAggregates/ContinuousAggregateDiffer.cs @@ -0,0 +1,113 @@ +using CmdScale.EntityFrameworkCore.TimescaleDB.Operations; +using Microsoft.EntityFrameworkCore.Metadata; +using Microsoft.EntityFrameworkCore.Migrations.Operations; + +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Internals.Features.ContinuousAggregates +{ + internal class ContinuousAggregateDiffer : IFeatureDiffer + { + public IReadOnlyList GetDifferences(IRelationalModel? source, IRelationalModel? target) + { + List operations = []; + + List sourceAggregates = [.. ContinuousAggregateModelExtractor.GetContinuousAggregates(source)]; + List targetAggregates = [.. ContinuousAggregateModelExtractor.GetContinuousAggregates(target)]; + + // Find new continuous aggregates - only compare by MaterializedViewName, not Schema + IEnumerable newAggregates = targetAggregates + .Where(t => !sourceAggregates.Any(s => s.MaterializedViewName == t.MaterializedViewName)); + operations.AddRange(newAggregates); + + // Find updated continuous aggregates + // Note: Only certain properties can be altered (ChunkInterval, CreateGroupIndexes, MaterializedOnly) + // For structural changes (time bucket, aggregates, group by, where), drop and recreate is required + var updatedAggregates = targetAggregates + .Join( + sourceAggregates, + target => (target.Schema, target.MaterializedViewName), + source => (source.Schema, source.MaterializedViewName), + (target, source) => new { Target = target, Source = source } + ) + .Where(x => + x.Target.ChunkInterval != x.Source.ChunkInterval || + x.Target.CreateGroupIndexes != x.Source.CreateGroupIndexes || + x.Target.MaterializedOnly != x.Source.MaterializedOnly + ); + + foreach (var aggregate in updatedAggregates) + { + operations.Add(new AlterContinuousAggregateOperation + { + Schema = aggregate.Target.Schema, + MaterializedViewName = aggregate.Target.MaterializedViewName, + ChunkInterval = aggregate.Target.ChunkInterval, + CreateGroupIndexes = aggregate.Target.CreateGroupIndexes, + MaterializedOnly = aggregate.Target.MaterializedOnly, + OldChunkInterval = aggregate.Source.ChunkInterval, + OldCreateGroupIndexes = aggregate.Source.CreateGroupIndexes, + OldMaterializedOnly = aggregate.Source.MaterializedOnly + }); + } + + // Find structural changes that require drop and recreate + var structurallyChangedAggregates = targetAggregates + .Join( + sourceAggregates, + target => (target.Schema, target.MaterializedViewName), + source => (source.Schema, source.MaterializedViewName), + (target, source) => new { Target = target, Source = source } + ) + .Where(x => + x.Target.ParentName != x.Source.ParentName || + x.Target.TimeBucketWidth != x.Source.TimeBucketWidth || + x.Target.TimeBucketSourceColumn != x.Source.TimeBucketSourceColumn || + x.Target.TimeBucketGroupBy != x.Source.TimeBucketGroupBy || + x.Target.WithNoData != x.Source.WithNoData || + !AreAggregateFunctionsEqual(x.Target.AggregateFunctions, x.Source.AggregateFunctions) || + !AreGroupByColumnsEqual(x.Target.GroupByColumns, x.Source.GroupByColumns) || + x.Target.WhereClaus != x.Source.WhereClaus + ); + + foreach (var aggregate in structurallyChangedAggregates) + { + operations.Add(new DropContinuousAggregateOperation + { + Schema = aggregate.Source.Schema, + MaterializedViewName = aggregate.Source.MaterializedViewName + }); + + operations.Add(aggregate.Target); + } + + // Find removed continuous aggregates + IEnumerable removedAggregates = sourceAggregates + .Where(s => !targetAggregates.Any(t => t.MaterializedViewName == s.MaterializedViewName)) + .Select(s => new DropContinuousAggregateOperation + { + Schema = s.Schema, + MaterializedViewName = s.MaterializedViewName + }); + operations.AddRange(removedAggregates); + + return operations; + } + + private static bool AreAggregateFunctionsEqual(List? list1, List? list2) + { + if (list1 == null && list2 == null) return true; + if (list1 == null || list2 == null) return false; + if (list1.Count != list2.Count) return false; + + return list1.SequenceEqual(list2); + } + + private static bool AreGroupByColumnsEqual(List? list1, List? list2) + { + if (list1 == null && list2 == null) return true; + if (list1 == null || list2 == null) return false; + if (list1.Count != list2.Count) return false; + + return list1.SequenceEqual(list2); + } + } +} diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB/Internals/Features/ContinuousAggregates/ContinuousAggregateModelExtractor.cs b/CmdScale.EntityFrameworkCore.TimescaleDB/Internals/Features/ContinuousAggregates/ContinuousAggregateModelExtractor.cs new file mode 100644 index 0000000..a7ee4e4 --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB/Internals/Features/ContinuousAggregates/ContinuousAggregateModelExtractor.cs @@ -0,0 +1,156 @@ +using CmdScale.EntityFrameworkCore.TimescaleDB.Configuration.ContinuousAggregate; +using CmdScale.EntityFrameworkCore.TimescaleDB.Operations; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; +using Microsoft.EntityFrameworkCore.Metadata; + +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Internals.Features.ContinuousAggregates +{ + internal class ContinuousAggregateModelExtractor + { + public static IEnumerable GetContinuousAggregates(IRelationalModel? relationalModel) + { + if (relationalModel == null) + { + yield break; + } + + foreach (IEntityType entityType in relationalModel.Model.GetEntityTypes()) + { + // Check if this entity is configured as a continuous aggregate + string? materializedViewName = entityType.FindAnnotation(ContinuousAggregateAnnotations.MaterializedViewName)?.Value as string; + if (string.IsNullOrWhiteSpace(materializedViewName)) + { + continue; + } + + // Get the parent (source) entity name + string? parentModelName = entityType.FindAnnotation(ContinuousAggregateAnnotations.ParentName)?.Value as string; + if (string.IsNullOrWhiteSpace(parentModelName)) + { + continue; + } + + // Find the parent entity type to get its table name + IEntityType? parentEntityType = relationalModel.Model.GetEntityTypes() + .FirstOrDefault(e => e.ClrType?.Name == parentModelName || e.ShortName() == parentModelName); + if (parentEntityType == null) + { + continue; + } + + string? parentTableName = parentEntityType.GetTableName(); + if (string.IsNullOrWhiteSpace(parentTableName)) + { + continue; + } + + // Get time bucket configuration + string? timeBucketWidth = entityType.FindAnnotation(ContinuousAggregateAnnotations.TimeBucketWidth)?.Value as string; + if (string.IsNullOrWhiteSpace(timeBucketWidth)) + { + continue; + } + + string? timeBucketSourceColumnModelName = entityType.FindAnnotation(ContinuousAggregateAnnotations.TimeBucketSourceColumn)?.Value as string; + if (string.IsNullOrWhiteSpace(timeBucketSourceColumnModelName)) + { + continue; + } + + // Get convention-aware store identifier for the parent table + StoreObjectIdentifier parentStoreIdentifier = StoreObjectIdentifier.Table(parentTableName, parentEntityType.GetSchema()); + + // Resolve time bucket source column to database column name + string? timeBucketSourceColumn = parentEntityType.FindProperty(timeBucketSourceColumnModelName)?.GetColumnName(parentStoreIdentifier); + if (string.IsNullOrWhiteSpace(timeBucketSourceColumn)) + { + continue; + } + + // Get optional configuration + bool timeBucketGroupBy = entityType.FindAnnotation(ContinuousAggregateAnnotations.TimeBucketGroupBy)?.Value as bool? ?? true; + string? chunkInterval = entityType.FindAnnotation(ContinuousAggregateAnnotations.ChunkInterval)?.Value as string; + bool withNoData = entityType.FindAnnotation(ContinuousAggregateAnnotations.WithNoData)?.Value as bool? ?? false; + bool createGroupIndexes = entityType.FindAnnotation(ContinuousAggregateAnnotations.CreateGroupIndexes)?.Value as bool? ?? false; + bool materializedOnly = entityType.FindAnnotation(ContinuousAggregateAnnotations.MaterializedOnly)?.Value as bool? ?? false; + string? whereClause = entityType.FindAnnotation(ContinuousAggregateAnnotations.WhereClause)?.Value as string; + + // Process aggregate functions - convert model property names to database column names + List aggregateFunctions = []; + IAnnotation? aggregateFunctionsAnnotation = entityType.FindAnnotation(ContinuousAggregateAnnotations.AggregateFunctions); + if (aggregateFunctionsAnnotation?.Value is List modelAggregateFunctions) + { + foreach (string aggInfo in modelAggregateFunctions) + { + string[] parts = aggInfo.Split(':'); + if (parts.Length != 3) + { + // Skip malformed string + continue; + } + + string aliasModelName = parts[0]; + string functionEnumString = parts[1]; + string sourceColumnModelName = parts[2]; + + // Resolve source column name from parent entity + string? sourceColumnDbName = parentEntityType.FindProperty(sourceColumnModelName)?.GetColumnName(parentStoreIdentifier); + if (string.IsNullOrWhiteSpace(sourceColumnDbName)) + { + // Skip if source column not found + continue; + } + + // Alias stays as-is since it's the target column name in the aggregate view + aggregateFunctions.Add($"{aliasModelName}:{functionEnumString}:{sourceColumnDbName}"); + } + } + + // Process group by columns - convert model property names to database column names + // Note: Some group by columns might be raw SQL expressions (e.g., "1, 2"), not property names + List groupByColumns = []; + IAnnotation? groupByColumnsAnnotation = entityType.FindAnnotation(ContinuousAggregateAnnotations.GroupByColumns); + if (groupByColumnsAnnotation?.Value is List modelGroupByColumns) + { + foreach (string modelColumn in modelGroupByColumns) + { + // Try to resolve as a property name from the parent entity + string? dbColumnName = parentEntityType.FindProperty(modelColumn)?.GetColumnName(parentStoreIdentifier); + + if (!string.IsNullOrWhiteSpace(dbColumnName)) + { + // It's a property name, use the resolved database column name + groupByColumns.Add(dbColumnName); + } + else + { + // It's not a property, assume it's a raw SQL expression and use as-is + groupByColumns.Add(modelColumn); + } + } + } + + // Use parent table's schema for the continuous aggregate + string schema = parentEntityType.GetSchema() ?? entityType.GetSchema() ?? DefaultValues.DefaultSchema; + + yield return new CreateContinuousAggregateOperation + { + Schema = schema, + MaterializedViewName = materializedViewName, + ParentName = parentTableName, + ChunkInterval = chunkInterval, + WithNoData = withNoData, + CreateGroupIndexes = createGroupIndexes, + MaterializedOnly = materializedOnly, + TimeBucketWidth = timeBucketWidth, + TimeBucketSourceColumn = timeBucketSourceColumn, + TimeBucketGroupBy = timeBucketGroupBy, + AggregateFunctions = aggregateFunctions, + GroupByColumns = groupByColumns, + WhereClaus = whereClause + }; + } + } + } +} diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB/Internals/Features/Hypertables/HypertableDiffer.cs b/CmdScale.EntityFrameworkCore.TimescaleDB/Internals/Features/Hypertables/HypertableDiffer.cs new file mode 100644 index 0000000..23d9838 --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB/Internals/Features/Hypertables/HypertableDiffer.cs @@ -0,0 +1,63 @@ +using CmdScale.EntityFrameworkCore.TimescaleDB.Operations; +using Microsoft.EntityFrameworkCore.Metadata; +using Microsoft.EntityFrameworkCore.Migrations.Operations; + +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Internals.Features.Hypertables +{ + internal class HypertableDiffer : IFeatureDiffer + { + public IReadOnlyList GetDifferences(IRelationalModel? source, IRelationalModel? target) + { + List operations = []; + + List sourceHypertables = [.. HypertableModelExtractor.GetHypertables(source)]; + List targetHypertables = [.. HypertableModelExtractor.GetHypertables(target)]; + + // Find new hypertables + IEnumerable newHypertables = targetHypertables.Where(t => !sourceHypertables.Any(s => s.TableName == t.TableName)); + operations.AddRange(newHypertables); + + // Find updated hypertables + var updatedHypertables = targetHypertables + .Join( + sourceHypertables, + target => (target.Schema, target.TableName), + source => (source.Schema, source.TableName), + (target, source) => new { Target = target, Source = source } + ) + .Where(x => + x.Target.ChunkTimeInterval != x.Source.ChunkTimeInterval || + x.Target.EnableCompression != x.Source.EnableCompression || + !AreChunkSkipColumnsEqual(x.Target.ChunkSkipColumns, x.Source.ChunkSkipColumns) + ); + + foreach (var hypertable in updatedHypertables) + { + operations.Add(new AlterHypertableOperation + { + TableName = hypertable.Target.TableName, + Schema = hypertable.Target.Schema, + ChunkTimeInterval = hypertable.Target.ChunkTimeInterval, + EnableCompression = hypertable.Target.EnableCompression, + ChunkSkipColumns = hypertable.Target.ChunkSkipColumns, + OldChunkTimeInterval = hypertable.Source.ChunkTimeInterval, + OldEnableCompression = hypertable.Source.EnableCompression, + OldChunkSkipColumns = hypertable.Source.ChunkSkipColumns + }); + } + + // TODO: Detect dropped hypertables if TimescaleDB supports a "de-hyper" operation. + + return operations; + } + + private static bool AreChunkSkipColumnsEqual(IReadOnlyList? list1, IReadOnlyList? list2) + { + if (list1 == null && list2 == null) return true; + if (list1 == null || list2 == null) return false; + if (list1.Count != list2.Count) return false; + + return new HashSet(list1).SetEquals(list2); + } + } +} \ No newline at end of file diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB/Internals/Features/Hypertables/HypertableModelExtractor.cs b/CmdScale.EntityFrameworkCore.TimescaleDB/Internals/Features/Hypertables/HypertableModelExtractor.cs new file mode 100644 index 0000000..bd35795 --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB/Internals/Features/Hypertables/HypertableModelExtractor.cs @@ -0,0 +1,93 @@ +using CmdScale.EntityFrameworkCore.TimescaleDB.Abstractions; +using CmdScale.EntityFrameworkCore.TimescaleDB.Configuration.Hypertable; +using CmdScale.EntityFrameworkCore.TimescaleDB.Operations; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; +using Microsoft.EntityFrameworkCore.Metadata; +using System.Text.Json; + +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Internals.Features.Hypertables +{ + internal static class HypertableModelExtractor + { + public static IEnumerable GetHypertables(IRelationalModel? relationalModel) + { + if (relationalModel == null) + { + yield break; + } + + foreach (IEntityType entityType in relationalModel.Model.GetEntityTypes()) + { + // Retrieve the annotations set by the convention + bool isHypertable = entityType.FindAnnotation(HypertableAnnotations.IsHypertable)?.Value as bool? ?? false; + if (!isHypertable) + { + continue; + } + + // Get convention-aware store identifier for the table + StoreObjectIdentifier storeIdentifier = StoreObjectIdentifier.Table(entityType.GetTableName()!, entityType.GetSchema()); + + string? timeColumnModelName = entityType.FindAnnotation(HypertableAnnotations.HypertableTimeColumn)?.Value as string; + if (string.IsNullOrWhiteSpace(timeColumnModelName)) + { + continue; + } + + string? timeColumnName = entityType.FindProperty(timeColumnModelName)?.GetColumnName(storeIdentifier); + if (string.IsNullOrWhiteSpace(timeColumnName)) + { + continue; + } + + + string? chunkSkipColumnsString = entityType.FindAnnotation(HypertableAnnotations.ChunkSkipColumns)?.Value as string; + List? chunkSkipColumns = null; + if (!string.IsNullOrWhiteSpace(chunkSkipColumnsString)) + { + chunkSkipColumns = chunkSkipColumnsString.Split(',', StringSplitOptions.TrimEntries) + .Select(modelPropName => entityType.FindProperty(modelPropName)?.GetColumnName(storeIdentifier)) + .Where(name => name != null) + .ToList()!; + } + + + List? additionalDimensions = null; + IAnnotation? additionalDimensionsAnnotations = entityType.FindAnnotation(HypertableAnnotations.AdditionalDimensions); + if (additionalDimensionsAnnotations?.Value is string json && !string.IsNullOrWhiteSpace(json)) + { + List? modelDimensions = JsonSerializer.Deserialize>(json); + if (modelDimensions != null) + { + additionalDimensions = []; + foreach (Dimension dim in modelDimensions) + { + string? conventionalColumnName = entityType.FindProperty(dim.ColumnName)?.GetColumnName(storeIdentifier); + if (conventionalColumnName != null) + { + Dimension newDimension = JsonSerializer.Deserialize(JsonSerializer.Serialize(dim))!; + newDimension.ColumnName = conventionalColumnName; + additionalDimensions.Add(newDimension); + } + } + } + } + + string chunkTimeInterval = entityType.FindAnnotation(HypertableAnnotations.ChunkTimeInterval)?.Value as string ?? DefaultValues.ChunkTimeInterval; + bool enableCompression = entityType.FindAnnotation(HypertableAnnotations.EnableCompression)?.Value as bool? ?? false; + + yield return new CreateHypertableOperation + { + TableName = entityType.GetTableName()!, + Schema = entityType.GetSchema() ?? DefaultValues.DefaultSchema, + TimeColumnName = timeColumnName, + ChunkTimeInterval = chunkTimeInterval ?? DefaultValues.ChunkTimeInterval, + EnableCompression = enableCompression, + ChunkSkipColumns = chunkSkipColumns, + AdditionalDimensions = additionalDimensions + }; + } + } + } +} \ No newline at end of file diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB/Internals/Features/IFeatureDiffer.cs b/CmdScale.EntityFrameworkCore.TimescaleDB/Internals/Features/IFeatureDiffer.cs new file mode 100644 index 0000000..e977a4a --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB/Internals/Features/IFeatureDiffer.cs @@ -0,0 +1,20 @@ +using Microsoft.EntityFrameworkCore.Metadata; +using Microsoft.EntityFrameworkCore.Migrations.Operations; + +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Internals.Features +{ + /// + /// Defines a contract for a component that can detect differences for a specific + /// TimescaleDB feature between two model states. + /// + public interface IFeatureDiffer + { + /// + /// Gets the migration operations needed to transition from the source to the target model. + /// + /// The source model (from the last migration). + /// The target model (the current state). + /// A collection of migration operations. + IReadOnlyList GetDifferences(IRelationalModel? source, IRelationalModel? target); + } +} diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB/Internals/Features/ReorderPolicies/ReorderPolicyDiffer.cs b/CmdScale.EntityFrameworkCore.TimescaleDB/Internals/Features/ReorderPolicies/ReorderPolicyDiffer.cs new file mode 100644 index 0000000..18ecc29 --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB/Internals/Features/ReorderPolicies/ReorderPolicyDiffer.cs @@ -0,0 +1,69 @@ +using CmdScale.EntityFrameworkCore.TimescaleDB.Operations; +using Microsoft.EntityFrameworkCore.Metadata; +using Microsoft.EntityFrameworkCore.Migrations.Operations; + +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Internals.Features.ReorderPolicies +{ + internal class ReorderPolicyDiffer : IFeatureDiffer + { + public IReadOnlyList GetDifferences(IRelationalModel? source, IRelationalModel? target) + { + // Get the standard migration operations (CreateTable, AddColumn, etc.) from the base MigrationsModelDiffer. + List operations = []; + + // Reorder diffs + List sourcePolicies = [.. ReorderPolicyModelExtractor.GetReorderPolicies(source)]; + List targetPolicies = [.. ReorderPolicyModelExtractor.GetReorderPolicies(target)]; + + // Identiy new reorder policies + IEnumerable newReorderPolicies = targetPolicies.Where(t => !sourcePolicies.Any(s => s.TableName == t.TableName)); + operations.AddRange(newReorderPolicies); + + // Identify updated reorder policies + var updatedReorderPolicies = targetPolicies + .Join( + sourcePolicies, + targetPolicy => (targetPolicy.Schema, targetPolicy.TableName), + sourcePolicy => (sourcePolicy.Schema, sourcePolicy.TableName), + (targetPolicy, sourcePolicy) => new { Target = targetPolicy, Source = sourcePolicy } + ) + .Where(x => + x.Target.IndexName != x.Source.IndexName || + x.Target.InitialStart != x.Source.InitialStart || + x.Target.ScheduleInterval != x.Source.ScheduleInterval || + x.Target.MaxRuntime != x.Source.MaxRuntime || + x.Target.MaxRetries != x.Source.MaxRetries || + x.Target.RetryPeriod != x.Source.RetryPeriod + ); + + foreach (var policy in updatedReorderPolicies) + { + operations.Add(new AlterReorderPolicyOperation + { + TableName = policy.Target.TableName, + Schema = policy.Target.Schema, + IndexName = policy.Target.IndexName, + InitialStart = policy.Target.InitialStart, + ScheduleInterval = policy.Target.ScheduleInterval, + MaxRuntime = policy.Target.MaxRuntime, + MaxRetries = policy.Target.MaxRetries, + RetryPeriod = policy.Target.RetryPeriod, + + OldIndexName = policy.Source.IndexName, + OldInitialStart = policy.Source.InitialStart, + OldScheduleInterval = policy.Source.ScheduleInterval, + OldMaxRuntime = policy.Source.MaxRuntime, + OldMaxRetries = policy.Source.MaxRetries, + OldRetryPeriod = policy.Source.RetryPeriod + }); + } + + IEnumerable removedReorderPolicies = sourcePolicies + .Where(s => !targetPolicies.Any(t => t.TableName == s.TableName && t.Schema == s.Schema)) + .Select(p => new DropReorderPolicyOperation { TableName = p.TableName, Schema = p.Schema }); + operations.AddRange(removedReorderPolicies); + + return operations; + } + } +} diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB/Internals/Features/ReorderPolicies/ReorderPolicyModelExtractor.cs b/CmdScale.EntityFrameworkCore.TimescaleDB/Internals/Features/ReorderPolicies/ReorderPolicyModelExtractor.cs new file mode 100644 index 0000000..c461ff5 --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB/Internals/Features/ReorderPolicies/ReorderPolicyModelExtractor.cs @@ -0,0 +1,48 @@ +using CmdScale.EntityFrameworkCore.TimescaleDB.Configuration.ReorderPolicy; +using CmdScale.EntityFrameworkCore.TimescaleDB.Operations; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Metadata; + +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Internals.Features.ReorderPolicies +{ + internal static class ReorderPolicyModelExtractor + { + public static IEnumerable GetReorderPolicies(IRelationalModel? relationalModel) + { + if (relationalModel == null) + { + yield break; + } + + foreach (IEntityType entityType in relationalModel.Model.GetEntityTypes()) + { + // Retrieve the annotations set by the convention + bool hasReorderPolicy = entityType.FindAnnotation(ReorderPolicyAnnotations.HasReorderPolicy)?.Value as bool? ?? false; + if (!hasReorderPolicy) + { + continue; + } + + string? indexName = entityType.FindAnnotation(ReorderPolicyAnnotations.IndexName)?.Value as string; + if (string.IsNullOrWhiteSpace(indexName)) + { + continue; + } + + DateTime? initialStart = entityType.FindAnnotation(ReorderPolicyAnnotations.InitialStart)?.Value as DateTime?; + + yield return new AddReorderPolicyOperation + { + TableName = entityType.GetTableName()!, + Schema = entityType.GetSchema() ?? DefaultValues.DefaultSchema, + IndexName = indexName!, + InitialStart = initialStart, + ScheduleInterval = entityType.FindAnnotation(ReorderPolicyAnnotations.ScheduleInterval)?.Value as string ?? DefaultValues.ReorderPolicyScheduleInterval, + MaxRuntime = entityType.FindAnnotation(ReorderPolicyAnnotations.MaxRuntime)?.Value as string ?? DefaultValues.ReorderPolicyMaxRuntime, + MaxRetries = entityType.FindAnnotation(ReorderPolicyAnnotations.MaxRetries)?.Value as int? ?? DefaultValues.ReorderPolicyMaxRetries, + RetryPeriod = entityType.FindAnnotation(ReorderPolicyAnnotations.RetryPeriod)?.Value as string ?? DefaultValues.ReorderPolicyRetryPeriod + }; + } + } + } +} diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB/Internals/TimescaleMigrationsModelDiffer.cs b/CmdScale.EntityFrameworkCore.TimescaleDB/Internals/TimescaleMigrationsModelDiffer.cs new file mode 100644 index 0000000..a3ebbb3 --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB/Internals/TimescaleMigrationsModelDiffer.cs @@ -0,0 +1,74 @@ +using CmdScale.EntityFrameworkCore.TimescaleDB.Internals.Features; +using CmdScale.EntityFrameworkCore.TimescaleDB.Internals.Features.ContinuousAggregates; +using CmdScale.EntityFrameworkCore.TimescaleDB.Internals.Features.Hypertables; +using CmdScale.EntityFrameworkCore.TimescaleDB.Internals.Features.ReorderPolicies; +using CmdScale.EntityFrameworkCore.TimescaleDB.Operations; +using Microsoft.EntityFrameworkCore.Metadata; +using Microsoft.EntityFrameworkCore.Migrations; +using Microsoft.EntityFrameworkCore.Migrations.Internal; +using Microsoft.EntityFrameworkCore.Migrations.Operations; +using Microsoft.EntityFrameworkCore.Storage; +using Microsoft.EntityFrameworkCore.Update.Internal; + +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Internals +{ +#pragma warning disable EF1001 // Suppress warning about internal APIs usage, common for providers/extensions + public class TimescaleMigrationsModelDiffer( + IRelationalTypeMappingSource typeMappingSource, + IMigrationsAnnotationProvider migrationsAnnotationProvider, + IRelationalAnnotationProvider relationalAnnotationProvider, + IRowIdentityMapFactory rowIdentityMapFactory, + CommandBatchPreparerDependencies commandBatchPreparerDependencies) : MigrationsModelDiffer(typeMappingSource, migrationsAnnotationProvider, relationalAnnotationProvider, rowIdentityMapFactory, commandBatchPreparerDependencies) + { + private readonly IReadOnlyList _featureDiffers = [ + new HypertableDiffer(), + new ReorderPolicyDiffer(), + new ContinuousAggregateDiffer(), + ]; + + public override IReadOnlyList GetDifferences(IRelationalModel? source, IRelationalModel? target) + { + // Get all operations + List allOperations = [.. base.GetDifferences(source, target)]; + + foreach (IFeatureDiffer differ in _featureDiffers) + { + allOperations.AddRange(differ.GetDifferences(source, target)); + } + + // Sort the entire list based on the priority defined in the helper method + allOperations.Sort((op1, op2) => GetOperationPriority(op1).CompareTo(GetOperationPriority(op2))); + + return allOperations; + } + + /// + /// Assigns a priority to operations to ensure correct execution order. + /// Lower numbers execute first. + /// + private static int GetOperationPriority(MigrationOperation operation) + { + switch (operation) + { + case CreateHypertableOperation: + return 10; + + case AddReorderPolicyOperation: + case AlterReorderPolicyOperation: + case DropReorderPolicyOperation: + return 20; + + case CreateContinuousAggregateOperation: + return 30; + case AlterContinuousAggregateOperation: + case DropContinuousAggregateOperation: + return 40; + + // Standard EF Core operations (CreateTable, etc.) + default: + return 0; + } + } + } +#pragma warning restore EF1001 +} \ No newline at end of file diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB/Internals/WhereClauseEpressionVisitor.cs b/CmdScale.EntityFrameworkCore.TimescaleDB/Internals/WhereClauseEpressionVisitor.cs new file mode 100644 index 0000000..775ba0e --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB/Internals/WhereClauseEpressionVisitor.cs @@ -0,0 +1,153 @@ +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Metadata; +using System.Linq.Expressions; +using System.Reflection; +using System.Text; + +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Internals +{ + + /// + /// A simplified visitor to translate a WHERE clause LambdaExpression into a SQL string. + /// This must be used by your IMigrationsModelDiffer, not the SQL generator. + /// + public class WhereClauseExpressionVisitor(IEntityType sourceEntityType) : ExpressionVisitor + { + private readonly StringBuilder sqlBuilder = new(); + + /// + /// Translates the provided expression into a SQL WHERE clause. + /// + public string Translate(Expression expression) + { + sqlBuilder.Clear(); + Visit(expression); + return sqlBuilder.ToString(); + } + + // Handles expressions like x.Property == "value" + protected override Expression VisitBinary(BinaryExpression node) + { + sqlBuilder.Append('('); + + // Visit the left side (e.g., x.Property) + Visit(node.Left); + + // Add the SQL operator + sqlBuilder.Append(GetSqlOperator(node.NodeType)); + + // Visit the right side (e.g., "value") + Visit(node.Right); + + sqlBuilder.Append(')'); + return node; + } + + // Handles member access, like x.Ticker + protected override Expression VisitMember(MemberExpression node) + { + // We only care about properties of the entity + if (node.Expression != null && node.Expression.NodeType == ExpressionType.Parameter) + { + if (node.Member is PropertyInfo propertyInfo) + { + // Find the property on the EF Core model + IProperty? property = sourceEntityType.FindProperty(propertyInfo.Name); + if (property != null) + { + StoreObjectIdentifier storeIdentifier = StoreObjectIdentifier.Table(sourceEntityType.GetTableName()!, sourceEntityType.GetSchema()); + // Get the *database column name* + string? columnName = property.GetColumnName(storeIdentifier); + // Append the quoted column name + sqlBuilder.Append($"\"{columnName}\""); + return node; + } + } + } + + // Fallback for other member access (e.g., accessing a variable) + // This will try to evaluate it and treat it as a constant. + if (TryEvaluate(node, out object? value)) + { + AppendSqlLiteral(value); + return node; + } + + throw new NotSupportedException($"Member access '{node.Member.Name}' is not supported."); + } + + // Handles constants, like "MCRS" or 100 + protected override Expression VisitConstant(ConstantExpression node) + { + AppendSqlLiteral(node.Value); + return node; + } + + // Tries to "compile" part of the expression to get its value + private static bool TryEvaluate(Expression expression, out object? value) + { + try + { + value = Expression.Lambda(expression).Compile().DynamicInvoke(); + return true; + } + catch + { + value = null; + return false; + } + } + + // Helper to format a .NET value as a SQL literal + private void AppendSqlLiteral(object? value) + { + if (value == null) + { + sqlBuilder.Append("NULL"); + } + else if (value is string str) + { + // Simple string quoting; for production, you might need more robust escaping + sqlBuilder.Append($"'{str.Replace("'", "''")}'"); + } + else if (value is bool b) + { + sqlBuilder.Append(b ? "TRUE" : "FALSE"); + } + else if (value is int || value is long || value is double || value is float || value is decimal) + { + sqlBuilder.Append(value.ToString()); + } + else + { + throw new NotSupportedException($"Value of type '{value.GetType().Name}' is not supported."); + } + } + + // Helper to map ExpressionType to a SQL operator + private static string GetSqlOperator(ExpressionType type) + { + switch (type) + { + case ExpressionType.Equal: + return " = "; + case ExpressionType.NotEqual: + return " <> "; + case ExpressionType.GreaterThan: + return " > "; + case ExpressionType.GreaterThanOrEqual: + return " >= "; + case ExpressionType.LessThan: + return " < "; + case ExpressionType.LessThanOrEqual: + return " <= "; + case ExpressionType.AndAlso: + return " AND "; + case ExpressionType.OrElse: + return " OR "; + default: + throw new NotSupportedException($"Operator '{type}' is not supported."); + } + } + } +} \ No newline at end of file diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB/Operations/AlterContinuousAggregateOperation.cs b/CmdScale.EntityFrameworkCore.TimescaleDB/Operations/AlterContinuousAggregateOperation.cs new file mode 100644 index 0000000..f04d87c --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB/Operations/AlterContinuousAggregateOperation.cs @@ -0,0 +1,19 @@ +using Microsoft.EntityFrameworkCore.Migrations.Operations; + +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Operations +{ + public class AlterContinuousAggregateOperation : MigrationOperation + { + public string Schema { get; set; } = string.Empty; + public string MaterializedViewName { get; set; } = string.Empty; + + public string? ChunkInterval { get; set; } + public string? OldChunkInterval { get; set; } + + public bool CreateGroupIndexes { get; set; } + public bool OldCreateGroupIndexes { get; set; } + + public bool MaterializedOnly { get; set; } + public bool OldMaterializedOnly { get; set; } + } +} diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB/Operations/CreateContinuousAggregateOperation.cs b/CmdScale.EntityFrameworkCore.TimescaleDB/Operations/CreateContinuousAggregateOperation.cs new file mode 100644 index 0000000..296d788 --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB/Operations/CreateContinuousAggregateOperation.cs @@ -0,0 +1,24 @@ +using Microsoft.EntityFrameworkCore.Migrations.Operations; + +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Operations +{ + public class CreateContinuousAggregateOperation : MigrationOperation + { + public string Schema { get; set; } = string.Empty; + public string MaterializedViewName { get; set; } = string.Empty; + public string ParentName { get; set; } = string.Empty; + public string? ChunkInterval { get; set; } + + public bool WithNoData { get; set; } + public bool CreateGroupIndexes { get; set; } + public bool MaterializedOnly { get; set; } + + public string TimeBucketWidth { get; set; } = string.Empty; + public string TimeBucketSourceColumn { get; set; } = string.Empty; + public bool TimeBucketGroupBy { get; set; } = true; + + public List AggregateFunctions { get; set; } = []; + public List GroupByColumns { get; set; } = []; + public string? WhereClaus { get; set; } + } +} diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB/Operations/DropContinuousAggregateOperation.cs b/CmdScale.EntityFrameworkCore.TimescaleDB/Operations/DropContinuousAggregateOperation.cs new file mode 100644 index 0000000..46199b8 --- /dev/null +++ b/CmdScale.EntityFrameworkCore.TimescaleDB/Operations/DropContinuousAggregateOperation.cs @@ -0,0 +1,10 @@ +using Microsoft.EntityFrameworkCore.Migrations.Operations; + +namespace CmdScale.EntityFrameworkCore.TimescaleDB.Operations +{ + public class DropContinuousAggregateOperation : MigrationOperation + { + public string Schema { get; set; } = string.Empty; + public string MaterializedViewName { get; set; } = string.Empty; + } +} diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB/TimescaleDbContextOptionsBuilderExtensions.cs b/CmdScale.EntityFrameworkCore.TimescaleDB/TimescaleDbContextOptionsBuilderExtensions.cs index d42e6c5..273bd12 100644 --- a/CmdScale.EntityFrameworkCore.TimescaleDB/TimescaleDbContextOptionsBuilderExtensions.cs +++ b/CmdScale.EntityFrameworkCore.TimescaleDB/TimescaleDbContextOptionsBuilderExtensions.cs @@ -1,5 +1,7 @@ -using CmdScale.EntityFrameworkCore.TimescaleDB.Configuration.Hypertable; +using CmdScale.EntityFrameworkCore.TimescaleDB.Configuration.ContinuousAggregate; +using CmdScale.EntityFrameworkCore.TimescaleDB.Configuration.Hypertable; using CmdScale.EntityFrameworkCore.TimescaleDB.Configuration.ReorderPolicy; +using CmdScale.EntityFrameworkCore.TimescaleDB.Internals; using Microsoft.EntityFrameworkCore; using Microsoft.EntityFrameworkCore.Infrastructure; using Microsoft.EntityFrameworkCore.Metadata.Conventions; @@ -50,7 +52,7 @@ private class TimescaleDbOptionsExtension : IDbContextOptionsExtension public void ApplyServices(IServiceCollection services) { - services.AddSingleton(); + services.AddSingleton(); services.AddScoped(); services.Replace(ServiceDescriptor.Scoped()); } @@ -70,12 +72,13 @@ private class ExtensionInfo(IDbContextOptionsExtension extension) : DbContextOpt } } - public class HypertableConventionSetPlugin : IConventionSetPlugin + public class TimescaleDbConventionSetPlugin : IConventionSetPlugin { public ConventionSet ModifyConventions(ConventionSet conventionSet) { conventionSet.EntityTypeAddedConventions.Add(new HypertableConvention()); conventionSet.EntityTypeAddedConventions.Add(new ReorderPolicyConvention()); + conventionSet.EntityTypeAddedConventions.Add(new ContinuousAggregateConvention()); return conventionSet; } } diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB/TimescaleDbMigrationsSqlGenerator.cs b/CmdScale.EntityFrameworkCore.TimescaleDB/TimescaleDbMigrationsSqlGenerator.cs index 2a82452..56a65e9 100644 --- a/CmdScale.EntityFrameworkCore.TimescaleDB/TimescaleDbMigrationsSqlGenerator.cs +++ b/CmdScale.EntityFrameworkCore.TimescaleDB/TimescaleDbMigrationsSqlGenerator.cs @@ -19,6 +19,8 @@ protected override void Generate( List statements; HypertableOperationGenerator? hypertableOperationGenerator = null; ReorderPolicyOperationGenerator? reorderPolicyOperationGenerator = null; + ContinuousAggregateOperationGenerator? continuousAggregateOperationGenerator = null; + bool suppressTransaction = false; switch (operation) { @@ -47,12 +49,28 @@ protected override void Generate( statements = reorderPolicyOperationGenerator.Generate(dropReorderPolicyOperation); break; + case CreateContinuousAggregateOperation createContinuousAggregateOperation: + continuousAggregateOperationGenerator ??= new(isDesignTime: false); + statements = continuousAggregateOperationGenerator.Generate(createContinuousAggregateOperation); + suppressTransaction = true; + break; + + case AlterContinuousAggregateOperation alterContinuousAggregateOperation: + continuousAggregateOperationGenerator ??= new(isDesignTime: false); + statements = continuousAggregateOperationGenerator.Generate(alterContinuousAggregateOperation); + break; + + case DropContinuousAggregateOperation dropContinuousAggregateOperation: + continuousAggregateOperationGenerator ??= new(isDesignTime: false); + statements = continuousAggregateOperationGenerator.Generate(dropContinuousAggregateOperation); + break; + default: base.Generate(operation, model, builder); return; } - SqlBuilderHelper.BuildQueryString(statements, builder); + SqlBuilderHelper.BuildQueryString(statements, builder, suppressTransaction); } } diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB/TimescaleDbServiceCollectionExtensions.cs b/CmdScale.EntityFrameworkCore.TimescaleDB/TimescaleDbServiceCollectionExtensions.cs index febcbc8..f5cbaf8 100644 --- a/CmdScale.EntityFrameworkCore.TimescaleDB/TimescaleDbServiceCollectionExtensions.cs +++ b/CmdScale.EntityFrameworkCore.TimescaleDB/TimescaleDbServiceCollectionExtensions.cs @@ -1,4 +1,5 @@ -using Microsoft.EntityFrameworkCore.Infrastructure; +using CmdScale.EntityFrameworkCore.TimescaleDB.Internals; +using Microsoft.EntityFrameworkCore.Infrastructure; using Microsoft.EntityFrameworkCore.Metadata.Conventions.Infrastructure; using Microsoft.EntityFrameworkCore.Migrations; using Microsoft.Extensions.DependencyInjection; @@ -14,7 +15,7 @@ public static IServiceCollection AddEntityFrameworkTimescaleDb(this IServiceColl { new EntityFrameworkRelationalServicesBuilder(services) .TryAdd() - .TryAdd(); + .TryAdd(); return services; } diff --git a/CmdScale.EntityFrameworkCore.TimescaleDB/TimescaleMigrationsModelDiffer.cs b/CmdScale.EntityFrameworkCore.TimescaleDB/TimescaleMigrationsModelDiffer.cs deleted file mode 100644 index c51967f..0000000 --- a/CmdScale.EntityFrameworkCore.TimescaleDB/TimescaleMigrationsModelDiffer.cs +++ /dev/null @@ -1,282 +0,0 @@ -using CmdScale.EntityFrameworkCore.TimescaleDB.Abstractions; -using CmdScale.EntityFrameworkCore.TimescaleDB.Configuration.Hypertable; -using CmdScale.EntityFrameworkCore.TimescaleDB.Configuration.ReorderPolicy; -using CmdScale.EntityFrameworkCore.TimescaleDB.Operations; -using Microsoft.EntityFrameworkCore; -using Microsoft.EntityFrameworkCore.Infrastructure; -using Microsoft.EntityFrameworkCore.Metadata; -using Microsoft.EntityFrameworkCore.Migrations; -using Microsoft.EntityFrameworkCore.Migrations.Internal; -using Microsoft.EntityFrameworkCore.Migrations.Operations; -using Microsoft.EntityFrameworkCore.Storage; -using Microsoft.EntityFrameworkCore.Update.Internal; -using System.Text.Json; - -namespace CmdScale.EntityFrameworkCore.TimescaleDB -{ -#pragma warning disable EF1001 // Suppress warning about internal APIs usage, common for providers/extensions - public class TimescaleMigrationsModelDiffer( - IRelationalTypeMappingSource typeMappingSource, - IMigrationsAnnotationProvider migrationsAnnotationProvider, - IRelationalAnnotationProvider relationalAnnotationProvider, - IRowIdentityMapFactory rowIdentityMapFactory, - CommandBatchPreparerDependencies commandBatchPreparerDependencies) : MigrationsModelDiffer( - typeMappingSource, - migrationsAnnotationProvider, - relationalAnnotationProvider, - rowIdentityMapFactory, - commandBatchPreparerDependencies) - { - public override IReadOnlyList GetDifferences(IRelationalModel? source, IRelationalModel? target) - { - // Get the standard migration operations (CreateTable, AddColumn, etc.) from the base MigrationsModelDiffer. - List operations = [.. base.GetDifferences(source, target)]; - - // Hypertable diffs - List targetHypertables = [.. GetHypertables(target)]; - List sourceHypertables = [.. GetHypertables(source)]; - - // Identify new hypertables - List newHypertables = [.. targetHypertables.Where(t => !sourceHypertables.Any(s => s.TableName == t.TableName))]; - - foreach (CreateHypertableOperation? hypertable in newHypertables) - { - int createTableOpIndex = operations.FindIndex(op => - op is CreateTableOperation createTable && - createTable.Name == hypertable.TableName); - - if (createTableOpIndex != -1) - { - operations.Insert(createTableOpIndex + 1, hypertable); - } - } - - // Identity updated hypertables - var updatedHypertables = targetHypertables - .Join( - sourceHypertables, - target => (target.Schema, target.TableName), - source => (source.Schema, source.TableName), - (target, source) => new { Target = target, Source = source } - ) - .Where(x => - x.Target.ChunkTimeInterval != x.Source.ChunkTimeInterval || - x.Target.EnableCompression != x.Source.EnableCompression || - !AreChunkSkipColumnsEqual(x.Target.ChunkSkipColumns, x.Source.ChunkSkipColumns) - ) - .ToList(); - - foreach (var hypertable in updatedHypertables) - { - AlterHypertableOperation alterOperation = new() - { - TableName = hypertable.Target.TableName, - Schema = hypertable.Target.Schema, - ChunkTimeInterval = hypertable.Target.ChunkTimeInterval, - EnableCompression = hypertable.Target.EnableCompression, - ChunkSkipColumns = hypertable.Target.ChunkSkipColumns, - - OldChunkTimeInterval = hypertable.Source.ChunkTimeInterval, - OldEnableCompression = hypertable.Source.EnableCompression, - OldChunkSkipColumns = hypertable.Source.ChunkSkipColumns - }; - - operations.Add(alterOperation); - } - - // Reorder diffs - List sourcePolicies = [.. GetReorderPolicies(source)]; - List targetPolicies = [.. GetReorderPolicies(target)]; - - // Identiy new reorder policies - IEnumerable newReorderPolicies = targetPolicies.Where(t => !sourcePolicies.Any(s => s.TableName == t.TableName)); - operations.AddRange(newReorderPolicies); - - // Identify updated reorder policies - var updatedReorderPolicies = targetPolicies - .Join( - sourcePolicies, - targetPolicy => (targetPolicy.Schema, targetPolicy.TableName), - sourcePolicy => (sourcePolicy.Schema, sourcePolicy.TableName), - (targetPolicy, sourcePolicy) => new { Target = targetPolicy, Source = sourcePolicy } - ) - .Where(x => - x.Target.IndexName != x.Source.IndexName || - x.Target.InitialStart != x.Source.InitialStart || - x.Target.ScheduleInterval != x.Source.ScheduleInterval || - x.Target.MaxRuntime != x.Source.MaxRuntime || - x.Target.MaxRetries != x.Source.MaxRetries || - x.Target.RetryPeriod != x.Source.RetryPeriod - ); - - foreach (var policy in updatedReorderPolicies) - { - operations.Add(new AlterReorderPolicyOperation - { - TableName = policy.Target.TableName, - Schema = policy.Target.Schema, - IndexName = policy.Target.IndexName, - InitialStart = policy.Target.InitialStart, - ScheduleInterval = policy.Target.ScheduleInterval, - MaxRuntime = policy.Target.MaxRuntime, - MaxRetries = policy.Target.MaxRetries, - RetryPeriod = policy.Target.RetryPeriod, - - OldIndexName = policy.Source.IndexName, - OldInitialStart = policy.Source.InitialStart, - OldScheduleInterval = policy.Source.ScheduleInterval, - OldMaxRuntime = policy.Source.MaxRuntime, - OldMaxRetries = policy.Source.MaxRetries, - OldRetryPeriod = policy.Source.RetryPeriod - }); - } - - IEnumerable removedReorderPolicies = sourcePolicies - .Where(s => !targetPolicies.Any(t => t.TableName == s.TableName)) - .Select(p => new DropReorderPolicyOperation { TableName = p.TableName }); - operations.AddRange(removedReorderPolicies); - - return operations; - } - - // Helper method to extract hypertable configuration from an IRelationalModel - private static IEnumerable GetHypertables(IRelationalModel? relationalModel) - { - if (relationalModel == null) - { - yield break; - } - - foreach (IEntityType entityType in relationalModel.Model.GetEntityTypes()) - { - // Retrieve the annotations set by the convention - bool isHypertable = entityType.FindAnnotation(HypertableAnnotations.IsHypertable)?.Value as bool? ?? false; - if (!isHypertable) - { - continue; - } - - // Get convention-aware store identifier for the table - StoreObjectIdentifier storeIdentifier = StoreObjectIdentifier.Table(entityType.GetTableName()!, entityType.GetSchema()); - - string? timeColumnModelName = entityType.FindAnnotation(HypertableAnnotations.HypertableTimeColumn)?.Value as string; - if (string.IsNullOrWhiteSpace(timeColumnModelName)) - { - continue; - } - - string? timeColumnName = entityType.FindProperty(timeColumnModelName)?.GetColumnName(storeIdentifier); - if (string.IsNullOrWhiteSpace(timeColumnName)) - { - continue; - } - - - string? chunkSkipColumnsString = entityType.FindAnnotation(HypertableAnnotations.ChunkSkipColumns)?.Value as string; - List? chunkSkipColumns = null; - if (!string.IsNullOrWhiteSpace(chunkSkipColumnsString)) - { - chunkSkipColumns = chunkSkipColumnsString.Split(',', StringSplitOptions.TrimEntries) - .Select(modelPropName => entityType.FindProperty(modelPropName)?.GetColumnName(storeIdentifier)) - .Where(name => name != null) - .ToList()!; - } - - - List? additionalDimensions = null; - IAnnotation? additionalDimensionsAnnotations = entityType.FindAnnotation(HypertableAnnotations.AdditionalDimensions); - if (additionalDimensionsAnnotations?.Value is string json && !string.IsNullOrWhiteSpace(json)) - { - List? modelDimensions = JsonSerializer.Deserialize>(json); - if (modelDimensions != null) - { - additionalDimensions = []; - foreach (Dimension dim in modelDimensions) - { - string? conventionalColumnName = entityType.FindProperty(dim.ColumnName)?.GetColumnName(storeIdentifier); - if (conventionalColumnName != null) - { - Dimension newDimension = JsonSerializer.Deserialize(JsonSerializer.Serialize(dim))!; - newDimension.ColumnName = conventionalColumnName; - additionalDimensions.Add(newDimension); - } - } - } - } - - string chunkTimeInterval = entityType.FindAnnotation(HypertableAnnotations.ChunkTimeInterval)?.Value as string ?? DefaultValues.ChunkTimeInterval; - bool enableCompression = entityType.FindAnnotation(HypertableAnnotations.EnableCompression)?.Value as bool? ?? false; - - yield return new CreateHypertableOperation - { - TableName = entityType.GetTableName()!, - Schema = entityType.GetSchema() ?? DefaultValues.DefaultSchema, - TimeColumnName = timeColumnName, - ChunkTimeInterval = chunkTimeInterval ?? DefaultValues.ChunkTimeInterval, - EnableCompression = enableCompression, - ChunkSkipColumns = chunkSkipColumns, - AdditionalDimensions = additionalDimensions - }; - } - } - - private static IEnumerable GetReorderPolicies(IRelationalModel? relationalModel) - { - if (relationalModel == null) - { - yield break; - } - - foreach (IEntityType entityType in relationalModel.Model.GetEntityTypes()) - { - // Retrieve the annotations set by the convention - bool hasReorderPolicy = entityType.FindAnnotation(ReorderPolicyAnnotations.HasReorderPolicy)?.Value as bool? ?? false; - if (!hasReorderPolicy) - { - continue; - } - - // Get convention-aware store identifier for the table - StoreObjectIdentifier storeIdentifier = StoreObjectIdentifier.Table(entityType.GetTableName()!, entityType.GetSchema()); - - string? indexModelName = entityType.FindAnnotation(ReorderPolicyAnnotations.IndexName)?.Value as string; - if (string.IsNullOrWhiteSpace(indexModelName)) - { - continue; - } - - string? indexName = entityType.FindIndex(indexModelName)?.GetDatabaseName(storeIdentifier); - if (string.IsNullOrWhiteSpace(indexName)) - { - continue; - } - - DateTime? initialStart = entityType.FindAnnotation(ReorderPolicyAnnotations.InitialStart)?.Value as DateTime?; - - yield return new AddReorderPolicyOperation - { - TableName = entityType.GetTableName()!, - Schema = entityType.GetSchema() ?? DefaultValues.DefaultSchema, - IndexName = indexName!, - InitialStart = initialStart, - ScheduleInterval = entityType.FindAnnotation(ReorderPolicyAnnotations.ScheduleInterval)?.Value as string ?? DefaultValues.ReorderPolicyScheduleInterval, - MaxRuntime = entityType.FindAnnotation(ReorderPolicyAnnotations.MaxRuntime)?.Value as string ?? DefaultValues.ReorderPolicyMaxRuntime, - MaxRetries = entityType.FindAnnotation(ReorderPolicyAnnotations.MaxRetries)?.Value as int? ?? DefaultValues.ReorderPolicyMaxRetries, - RetryPeriod = entityType.FindAnnotation(ReorderPolicyAnnotations.RetryPeriod)?.Value as string ?? DefaultValues.ReorderPolicyRetryPeriod - }; - } - } - - // Helper method to compare two lists of chunk skip columns - private static bool AreChunkSkipColumnsEqual(IReadOnlyList? list1, IReadOnlyList? list2) - { - if (list1 == null && list2 == null) return true; - if (list1 == null || list2 == null) return false; - if (list1.Count != list2.Count) return false; - - HashSet set1 = [.. list1]; - return set1.SetEquals(list2); - } - } -#pragma warning restore EF1001 -} \ No newline at end of file