Skip to content

Feature/continuous aggregate policies #34

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Merged
sebastian-ederer merged 9 commits into from
Feb 5, 2026
Merged

Feature/continuous aggregate policies #34

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
2aac8d6
Implements the support for continuous aggregate policies
sebastian-ederer Jan 2, 2026
ffb5c65
implements example for the continuous aggregate policies
sebastian-ederer Jan 13, 2026
445b2ac
test: Add comprehensive test coverage for continuous aggregate policies
sebastian-ederer Jan 30, 2026
f6ab044
test: Add comprehensive test coverage for continuous aggregate policies
sebastian-ederer Jan 30, 2026
53c6021
Ahmend merge
sebastian-ederer Feb 5, 2026
c46f9ca
Ahmend merge
sebastian-ederer Feb 5, 2026
0338cbe
Merge branch 'feature/continuous_aggregate_policies' of https://githu…
sebastian-ederer Feb 5, 2026
48ad607
Merge branch 'feature/continuous_aggregate_policies' of https://githu…
sebastian-ederer Feb 5, 2026
a019ee6
Merge branch 'feature/continuous_aggregate_policies' of https://githu…
sebastian-ederer Feb 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .gitignore
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -436,6 +436,7 @@ samples/Eftdb.Samples.DatabaseFirst/**/*.cs
# AI
CLAUDE.md
.claude
tmpclaude-*

# Code coverage report
coverage
Expand Down
38 changes: 19 additions & 19 deletions samples/Eftdb.Samples.DatabaseFirst/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,30 +1,30 @@
# EF Core Database-First Example with TimescaleDB
# EF Core Database-First Example with TimescaleDB

This project demonstrates how to use the **Database-First** approach with [TimescaleDB](https://www.timescale.com/) using the `CmdScale.EntityFrameworkCore.TimescaleDB` package.

---

## 📦 Required NuGet Packages
## Required NuGet Packages

Ensure the following package is installed in your project:

- `CmdScale.EntityFrameworkCore.TimescaleDB.Design`

---

## 🛠️ Scaffold DbContext and Models
## Scaffold DbContext and Models

Use the following command to scaffold the `DbContext` and entity classes from an existing TimescaleDB database:

```bash
dotnet ef dbcontext scaffold
"Host=localhost;Database=cmdscale-ef-timescaledb;Username=timescale_admin;Password=R#!kro#GP43ra8Ae"
CmdScale.EntityFrameworkCore.TimescaleDB.Design
--output-dir Models
--schema public
--context-dir .
--context MyTimescaleDbContext
--project CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess.DbFirst
dotnet ef dbcontext scaffold \
"Host=localhost;Database=cmdscale-ef-timescaledb;Username=timescale_admin;Password=R#!kro#GP43ra8Ae" \
CmdScale.EntityFrameworkCore.TimescaleDB.Design \
--output-dir Models \
--schema public \
--context-dir . \
--context MyTimescaleDbContext \
--project samples/Eftdb.Samples.DatabaseFirst
```

This command will:
Expand All @@ -37,20 +37,20 @@ This command will:

---

## 📁 Project Structure
## Project Structure

```text
CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess.DbFirst/
│
├── Models/ # Auto-generated entity models
└── MyTimescaleDbContext.cs # Auto-generated DbContext
samples/Eftdb.Samples.DatabaseFirst/
|
+-- Models/ # Auto-generated entity models
+-- MyTimescaleDbContext.cs # Auto-generated DbContext
```

---

## 🐳 Docker
## Docker

- A `docker-compose.yml` file is available in the **Solution Items** to spin up a TimescaleDB container for local development:
- A `docker-compose.yml` file is available at the repository root to spin up a TimescaleDB container for local development:

```bash
docker-compose up -d
Expand All @@ -60,7 +60,7 @@ CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess.DbFirst/

---

## 📚 Resources
## Resources

- [Entity Framework Core Documentation](https://learn.microsoft.com/en-us/ef/core/)
- [TimescaleDB Documentation](https://docs.timescale.com/)
5 changes: 4 additions & 1 deletion samples/Eftdb.Samples.Shared/Configurations/TradeAggregateConfiguration.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
using CmdScale.EntityFrameworkCore.TimescaleDB.Abstractions;
using CmdScale.EntityFrameworkCore.TimescaleDB.Configuration.ContinuousAggregate;
using CmdScale.EntityFrameworkCore.TimescaleDB.Configuration.ContinuousAggregatePolicy;
using CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess.Models;
using Microsoft.EntityFrameworkCore;
using Microsoft.EntityFrameworkCore.Metadata.Builders;
Expand All @@ -18,7 +19,9 @@ public void Configure(EntityTypeBuilder<TradeAggregate> builder)
.AddGroupByColumn(x => x.Exchange)
.AddGroupByColumn("1, 2")
.Where("\"ticker\" = 'MCRS'")
.MaterializedOnly();
.MaterializedOnly()
.WithRefreshPolicy(startOffset: "7 days", endOffset: "1 hour", scheduleInterval: "1 hour")
.WithRefreshNewestFirst(true);
}
}
}
6 changes: 6 additions & 0 deletions samples/Eftdb.Samples.Shared/Models/WeatherAggregate.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
using CmdScale.EntityFrameworkCore.TimescaleDB.Abstractions;
using CmdScale.EntityFrameworkCore.TimescaleDB.Configuration.ContinuousAggregate;
using CmdScale.EntityFrameworkCore.TimescaleDB.Configuration.ContinuousAggregatePolicy;
using Microsoft.EntityFrameworkCore;

namespace CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess.Models
Expand All @@ -18,6 +19,11 @@ namespace CmdScale.EntityFrameworkCore.TimescaleDB.Example.DataAccess.Models
MaterializedOnly = false,
Where = "\"temperature\" > -50 AND \"humidity\" >= 0")]
[TimeBucket("1 day", nameof(WeatherData.Time), GroupBy = true)]
[ContinuousAggregatePolicy(
StartOffset = "30 days",
EndOffset = "1 day",
ScheduleInterval = "1 hour",
RefreshNewestFirst = true)]
public class WeatherAggregate
{
// Avg aggregate function
Expand Down
70 changes: 70 additions & 0 deletions src/Eftdb.Design/Scaffolding/ContinuousAggregatePolicyAnnotationApplier.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,70 @@
using CmdScale.EntityFrameworkCore.TimescaleDB.Configuration.ContinuousAggregatePolicy;
using Microsoft.EntityFrameworkCore.Scaffolding.Metadata;
using static CmdScale.EntityFrameworkCore.TimescaleDB.Design.Scaffolding.ContinuousAggregatePolicyScaffoldingExtractor;

namespace CmdScale.EntityFrameworkCore.TimescaleDB.Design.Scaffolding
{
/// <summary>
/// Applies continuous aggregate policy annotations to scaffolded database views.
/// </summary>
public sealed class ContinuousAggregatePolicyAnnotationApplier : IAnnotationApplier
{
public void ApplyAnnotations(DatabaseTable table, object featureInfo)
{
if (featureInfo is not ContinuousAggregatePolicyInfo info)
{
throw new ArgumentException($"Expected {nameof(ContinuousAggregatePolicyInfo)}, got {featureInfo.GetType().Name}", nameof(featureInfo));
}

// Mark that this continuous aggregate has a refresh policy
table[ContinuousAggregatePolicyAnnotations.HasRefreshPolicy] = true;

// Apply start_offset and end_offset
if (!string.IsNullOrWhiteSpace(info.StartOffset))
{
table[ContinuousAggregatePolicyAnnotations.StartOffset] = info.StartOffset;
}

if (!string.IsNullOrWhiteSpace(info.EndOffset))
{
table[ContinuousAggregatePolicyAnnotations.EndOffset] = info.EndOffset;
}

// Apply schedule_interval
if (!string.IsNullOrWhiteSpace(info.ScheduleInterval))
{
table[ContinuousAggregatePolicyAnnotations.ScheduleInterval] = info.ScheduleInterval;
}

// Apply initial_start
if (info.InitialStart.HasValue)
{
table[ContinuousAggregatePolicyAnnotations.InitialStart] = info.InitialStart.Value;
}

// Apply include_tiered_data (only if not null - it's an optional parameter)
if (info.IncludeTieredData.HasValue)
{
table[ContinuousAggregatePolicyAnnotations.IncludeTieredData] = info.IncludeTieredData.Value;
}

// Apply buckets_per_batch (only if different from default value of 1)
if (info.BucketsPerBatch.HasValue && info.BucketsPerBatch.Value != 1)
{
table[ContinuousAggregatePolicyAnnotations.BucketsPerBatch] = info.BucketsPerBatch.Value;
}

// Apply max_batches_per_execution (only if different from default value of 0)
if (info.MaxBatchesPerExecution.HasValue && info.MaxBatchesPerExecution.Value != 0)
{
table[ContinuousAggregatePolicyAnnotations.MaxBatchesPerExecution] = info.MaxBatchesPerExecution.Value;
}

// Apply refresh_newest_first (only if different from default value of true)
if (info.RefreshNewestFirst.HasValue && !info.RefreshNewestFirst.Value)
{
table[ContinuousAggregatePolicyAnnotations.RefreshNewestFirst] = info.RefreshNewestFirst.Value;
}
}
}
}
141 changes: 141 additions & 0 deletions src/Eftdb.Design/Scaffolding/ContinuousAggregatePolicyScaffoldingExtractor.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,141 @@
using System.Data;
using System.Data.Common;
using System.Text.Json;

namespace CmdScale.EntityFrameworkCore.TimescaleDB.Design.Scaffolding
{
/// <summary>
/// Extracts continuous aggregate policy metadata from a TimescaleDB database for scaffolding.
/// </summary>
public sealed class ContinuousAggregatePolicyScaffoldingExtractor : ITimescaleFeatureExtractor
{
public sealed record ContinuousAggregatePolicyInfo(
string? StartOffset,
string? EndOffset,
string? ScheduleInterval,
DateTime? InitialStart,
bool? IncludeTieredData,
int? BucketsPerBatch,
int? MaxBatchesPerExecution,
bool? RefreshNewestFirst
);

public Dictionary<(string Schema, string TableName), object> Extract(DbConnection connection)
{
bool wasOpen = connection.State == ConnectionState.Open;
if (!wasOpen)
{
connection.Open();
}

try
{
Dictionary<(string, string), ContinuousAggregatePolicyInfo> policies = [];

using (DbCommand command = connection.CreateCommand())
{
// Query continuous aggregate policies from TimescaleDB jobs table
command.CommandText = @"
SELECT
ca.user_view_schema,
ca.user_view_name,
j.config,
j.schedule_interval::text,
j.initial_start
FROM timescaledb_information.jobs j
INNER JOIN _timescaledb_catalog.continuous_agg ca
ON (j.config->>'mat_hypertable_id')::integer = ca.mat_hypertable_id
WHERE j.proc_name = 'policy_refresh_continuous_aggregate';";

using DbDataReader reader = command.ExecuteReader();
while (reader.Read())
{
string viewSchema = reader.GetString(0);
string viewName = reader.GetString(1);
string? configJson = reader.IsDBNull(2) ? null : reader.GetString(2);
string? scheduleInterval = reader.IsDBNull(3) ? null : reader.GetString(3);
DateTime? initialStart = reader.IsDBNull(4) ? null : reader.GetDateTime(4);

// Parse the JSONB config to extract policy parameters
string? startOffset = null;
string? endOffset = null;
bool? includeTieredData = null;
int? bucketsPerBatch = null;
int? maxBatchesPerExecution = null;
bool? refreshNewestFirst = null;

if (!string.IsNullOrWhiteSpace(configJson))
{
using JsonDocument doc = JsonDocument.Parse(configJson);
JsonElement root = doc.RootElement;

// Extract start_offset
if (root.TryGetProperty("start_offset", out JsonElement startOffsetElement))
{
startOffset = IntervalParsingHelper.ParseIntervalOrInteger(startOffsetElement);
}

// Extract end_offset
if (root.TryGetProperty("end_offset", out JsonElement endOffsetElement))
{
endOffset = IntervalParsingHelper.ParseIntervalOrInteger(endOffsetElement);
}

// Extract include_tiered_data (optional)
if (root.TryGetProperty("include_tiered_data", out JsonElement includeTieredDataElement)
&& (includeTieredDataElement.ValueKind == JsonValueKind.True || includeTieredDataElement.ValueKind == JsonValueKind.False))
{
includeTieredData = includeTieredDataElement.GetBoolean();
}

// Extract buckets_per_batch (optional, defaults to 1)
if (root.TryGetProperty("buckets_per_batch", out JsonElement bucketsPerBatchElement)
&& bucketsPerBatchElement.ValueKind == JsonValueKind.Number)
{
bucketsPerBatch = bucketsPerBatchElement.GetInt32();
}

// Extract max_batches_per_execution (optional, defaults to 0)
if (root.TryGetProperty("max_batches_per_execution", out JsonElement maxBatchesElement)
&& maxBatchesElement.ValueKind == JsonValueKind.Number)
{
maxBatchesPerExecution = maxBatchesElement.GetInt32();
}

// Extract refresh_newest_first (optional, defaults to true)
if (root.TryGetProperty("refresh_newest_first", out JsonElement refreshNewestFirstElement)
&& (refreshNewestFirstElement.ValueKind == JsonValueKind.True || refreshNewestFirstElement.ValueKind == JsonValueKind.False))
{
refreshNewestFirst = refreshNewestFirstElement.GetBoolean();
}
}

policies[(viewSchema, viewName)] = new ContinuousAggregatePolicyInfo(
StartOffset: startOffset,
EndOffset: endOffset,
ScheduleInterval: scheduleInterval,
InitialStart: initialStart,
IncludeTieredData: includeTieredData,
BucketsPerBatch: bucketsPerBatch,
MaxBatchesPerExecution: maxBatchesPerExecution,
RefreshNewestFirst: refreshNewestFirst
);
}
}

// Convert to object dictionary to match interface
return policies.ToDictionary(
kvp => kvp.Key,
kvp => (object)kvp.Value
);
}
finally
{
if (!wasOpen)
{
connection.Close();
}
}
}
}
}
Loading