dev: SLO Entity Framework (#432)

Author: KirillKurdyukov

.github/workflows/slo-topic.yaml

@@ -26,11 +26,11 @@ jobs:
       - name: Prepare SLO Database
         run: |
           cd slo/src/TopicService
-          dotnet run create grpc://localhost:2135 /Root/testdb
+          dotnet run create "Host=localhost;Port=2135;Database=/Root/testdb"
       - name: Run SLO Tests
         continue-on-error: true
         run: |
           cd slo/src/TopicService
-          dotnet run run grpc://localhost:2135 /Root/testdb \
+          dotnet run run "Host=localhost;Port=2135;Database=/Root/testdb" \
             --write-rps 50 \
             --time 600

.github/workflows/slo.yml

@@ -31,6 +31,7 @@ jobs:
       matrix:
         workload:
           - AdoNet
+          - EF
 
     concurrency:
       group: slo-${{ github.ref }}-${{ matrix.workload }}
@@ -56,12 +57,12 @@ jobs:
       - name: Prepare SLO Database
         run: |
           cd slo/src/${{ matrix.workload }}
-          dotnet run create grpc://localhost:2135 /Root/testdb
+          dotnet run create "Host=localhost;Port=2135;Database=/Root/testdb"
 
       - name: Run SLO Tests
         run: |
           cd slo/src/${{ matrix.workload }}
-          dotnet run run grpc://localhost:2135 /Root/testdb \
+          dotnet run run "Host=localhost;Port=2135;Database=/Root/testdb" \
             --prom-pgw http://localhost:9091  \
             --report-period 250 \
             --time ${{inputs.slo_workload_duration_seconds || 600 }} \

slo/playground/configs/compose.yaml

@@ -5,7 +5,7 @@ x-runtime: &runtime
   network_mode: host
 
 x-ydb-node: &ydb-node
-  image: cr.yandex/crptqonuodf51kdj7a7d/ydb:24.3.11.13
+  image: cr.yandex/crptqonuodf51kdj7a7d/ydb:24.4.4.2
   restart: always
   <<: *runtime
   volumes:

slo/src/AdoNet/SloTableContext.cs

@@ -1,10 +1,10 @@
+using System.Data;
 using Internal;
 using Microsoft.Extensions.Logging;
 using Polly;
 using Prometheus;
 using Ydb.Sdk;
 using Ydb.Sdk.Ado;
-using Ydb.Sdk.Value;
 
 namespace AdoNet;
 
@@ -22,17 +22,36 @@ public class SloTableContext : SloTableContext<YdbDataSource>
 
     protected override string Job => "AdoNet";
 
-    protected override async Task Create(YdbDataSource client, string createTableSql, int operationTimeout)
+    protected override YdbDataSource CreateClient(Config config) => new(
+        new YdbConnectionStringBuilder(config.ConnectionString) { LoggerFactory = ISloContext.Factory }
+    );
+
+    protected override async Task Create(YdbDataSource client, int operationTimeout)
     {
         await using var ydbConnection = await client.OpenConnectionAsync();
-
         await new YdbCommand(ydbConnection)
-                { CommandText = createTableSql, CommandTimeout = operationTimeout }
-            .ExecuteNonQueryAsync();
+        {
+            CommandText = $"""
+                           CREATE TABLE `{SloTable.Name}` (
+                               Guid             UUID,
+                               Id               Int32,
+                               PayloadStr       Text,
+                               PayloadDouble    Double,
+                               PayloadTimestamp Timestamp,
+                               PRIMARY KEY (Guid, Id)
+                           );
+                           {SloTable.Options}
+                           """,
+            CommandTimeout = operationTimeout
+        }.ExecuteNonQueryAsync();
     }
 
-    protected override async Task<(int, StatusCode)> Upsert(YdbDataSource dataSource, string upsertSql,
-        Dictionary<string, YdbValue> parameters, int writeTimeout, Counter? errorsTotal = null)
+    protected override async Task<(int, StatusCode)> Save(
+        YdbDataSource client,
+        SloTable sloTable,
+        int writeTimeout,
+        Counter? errorsTotal = null
+    )
     {
         var context = new Context();
         if (errorsTotal != null)
@@ -42,15 +61,49 @@ protected override async Task Create(YdbDataSource client, string createTableSql
 
         var policyResult = await _policy.ExecuteAndCaptureAsync(async _ =>
         {
-            await using var ydbConnection = await dataSource.OpenConnectionAsync();
+            await using var ydbConnection = await client.OpenConnectionAsync();
 
             var ydbCommand = new YdbCommand(ydbConnection)
-                { CommandText = upsertSql, CommandTimeout = writeTimeout };
-
-            foreach (var (key, value) in parameters)
             {
-                ydbCommand.Parameters.AddWithValue(key, value);
-            }
+                CommandText = $"""
+                               INSERT INTO `{SloTable.Name}` (Guid, Id, PayloadStr, PayloadDouble, PayloadTimestamp)
+                               VALUES (@Guid, @Id, @PayloadStr, @PayloadDouble, @PayloadTimestamp)
+                               """,
+                CommandTimeout = writeTimeout,
+                Parameters =
+                {
+                    new YdbParameter
+                    {
+                        DbType = DbType.Guid,
+                        ParameterName = "Guid",
+                        Value = sloTable.Guid
+                    },
+                    new YdbParameter
+                    {
+                        DbType = DbType.Int32,
+                        ParameterName = "Id",
+                        Value = sloTable.Id
+                    },
+                    new YdbParameter
+                    {
+                        DbType = DbType.String,
+                        ParameterName = "PayloadStr",
+                        Value = sloTable.PayloadStr
+                    },
+                    new YdbParameter
+                    {
+                        DbType = DbType.Double,
+                        ParameterName = "PayloadDouble",
+                        Value = sloTable.PayloadDouble
+                    },
+                    new YdbParameter
+                    {
+                        DbType = DbType.DateTime2,
+                        ParameterName = "PayloadTimestamp",
+                        Value = sloTable.PayloadTimestamp
+                    }
+                }
+            };
 
             await ydbCommand.ExecuteNonQueryAsync();
         }, context);
@@ -60,8 +113,12 @@ protected override async Task Create(YdbDataSource client, string createTableSql
             ((YdbException)policyResult.FinalException)?.Code ?? StatusCode.Success);
     }
 
-    protected override async Task<(int, StatusCode, object?)> Select(YdbDataSource dataSource, string selectSql,
-        Dictionary<string, YdbValue> parameters, int readTimeout, Counter? errorsTotal = null)
+    protected override async Task<(int, StatusCode, object?)> Select(
+        YdbDataSource client,
+        (Guid Guid, int Id) select,
+        int readTimeout,
+        Counter? errorsTotal = null
+    )
     {
         var context = new Context();
         if (errorsTotal != null)
@@ -73,39 +130,33 @@ protected override async Task Create(YdbDataSource client, string createTableSql
         var policyResult = await _policy.ExecuteAndCaptureAsync(async _ =>
         {
             attempts++;
-            await using var ydbConnection = await dataSource.OpenConnectionAsync();
+            await using var ydbConnection = await client.OpenConnectionAsync();
 
             var ydbCommand = new YdbCommand(ydbConnection)
-                { CommandText = selectSql, CommandTimeout = readTimeout };
-
-            foreach (var (key, value) in parameters)
             {
-                ydbCommand.Parameters.AddWithValue(key, value);
-            }
+                CommandText = $"""
+                               SELECT Guid, Id, PayloadStr, PayloadDouble, PayloadTimestamp
+                               FROM `{SloTable.Name}` WHERE Guid = @Guid AND Id = @Id;
+                               """,
+                CommandTimeout = readTimeout,
+                Parameters =
+                {
+                    new YdbParameter { ParameterName = "Guid", DbType = DbType.Guid, Value = select.Guid },
+                    new YdbParameter { ParameterName = "Id", DbType = DbType.Int32, Value = select.Id }
+                }
+            };
 
             return await ydbCommand.ExecuteScalarAsync();
         }, context);
 
         return (attempts, ((YdbException)policyResult.FinalException)?.Code ?? StatusCode.Success, policyResult.Result);
     }
 
-    protected override Task<YdbDataSource> CreateClient(Config config)
+    protected override async Task<int> SelectCount(YdbDataSource client)
     {
-        var splitEndpoint = config.Endpoint.Split("://");
-        var useTls = splitEndpoint[0] switch
-        {
-            "grpc" => false,
-            "grpcs" => true,
-            _ => throw new ArgumentException("Don't support schema: " + splitEndpoint[0])
-        };
-
-        var host = splitEndpoint[1].Split(":")[0];
-        var port = splitEndpoint[1].Split(":")[1];
+        await using var ydbConnection = await client.OpenConnectionAsync();
 
-        return Task.FromResult(new YdbDataSource(new YdbConnectionStringBuilder
-        {
-            UseTls = useTls, Host = host, Port = int.Parse(port), Database = config.Db,
-            LoggerFactory = ISloContext.Factory
-        }));
+        return (int)(await new YdbCommand(ydbConnection) { CommandText = $"SELECT MAX(Id) FROM {SloTable.Name}" }
+            .ExecuteScalarAsync())!;
     }
 }

slo/src/EF/EF.csproj

@@ -0,0 +1,24 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <OutputType>Exe</OutputType>
+        <TargetFramework>net8.0</TargetFramework>
+        <ImplicitUsings>enable</ImplicitUsings>
+        <Nullable>enable</Nullable>
+    </PropertyGroup>
+
+    <ItemGroup>
+        <ProjectReference Include="..\Internal\Internal.csproj" />
+    </ItemGroup>
+
+    <ItemGroup>
+        <ProjectReference Include="..\..\..\src\EFCore.Ydb\src\EntityFrameworkCore.Ydb.csproj" />
+    </ItemGroup>
+
+    <ItemGroup>
+      <PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="9.0.5">
+        <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+        <PrivateAssets>all</PrivateAssets>
+      </PackageReference>
+    </ItemGroup>
+</Project>

slo/src/TableService/Program.cs renamed to slo/src/EF/Program.cs

@@ -1,6 +1,6 @@
 // See https://aka.ms/new-console-template for more information
 
+using EF;
 using Internal;
-using TableService;
 
-return await Cli.Run(new SloTableContext(), args);
+await Cli.Run(new SloTableContext(), args);

slo/src/EF/SloTableContext.cs

@@ -0,0 +1,59 @@
+using EntityFrameworkCore.Ydb.Extensions;
+using Internal;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Infrastructure;
+using Prometheus;
+using Ydb.Sdk;
+
+namespace EF;
+
+public class SloTableContext : SloTableContext<PooledDbContextFactory<TableDbContext>>
+{
+    protected override string Job => "EF";
+
+    protected override PooledDbContextFactory<TableDbContext> CreateClient(Config config) =>
+        new(new DbContextOptionsBuilder<TableDbContext>().UseYdb(config.ConnectionString).Options);
+
+    protected override async Task Create(
+        PooledDbContextFactory<TableDbContext> client,
+        int operationTimeout
+    )
+    {
+        await using var dbContext = await client.CreateDbContextAsync();
+        await dbContext.Database.EnsureCreatedAsync();
+        await dbContext.Database.ExecuteSqlRawAsync(SloTable.Options);
+    }
+
+    protected override async Task<(int, StatusCode)> Save(
+        PooledDbContextFactory<TableDbContext> client,
+        SloTable sloTable,
+        int writeTimeout,
+        Counter? errorsTotal = null)
+    {
+        await using var dbContext = await client.CreateDbContextAsync();
+        dbContext.SloEntities.Add(sloTable);
+        await dbContext.SaveChangesAsync();
+
+        return (1, StatusCode.Success);
+    }
+
+    protected override async Task<(int, StatusCode, object?)> Select(
+        PooledDbContextFactory<TableDbContext> client,
+        (Guid Guid, int Id) select,
+        int readTimeout,
+        Counter? errorsTotal = null
+    )
+    {
+        await using var dbContext = await client.CreateDbContextAsync();
+        await dbContext.SloEntities.FindAsync(select.Guid, select.Id);
+
+        return (0, StatusCode.Success, null);
+    }
+
+    protected override async Task<int> SelectCount(PooledDbContextFactory<TableDbContext> client)
+    {
+        await using var dbContext = await client.CreateDbContextAsync();
+
+        return await dbContext.SloEntities.CountAsync();
+    }
+}

slo/src/EF/TableDbContext.cs

@@ -0,0 +1,14 @@
+using Internal;
+using Microsoft.EntityFrameworkCore;
+
+namespace EF;
+
+public class TableDbContext(DbContextOptions<TableDbContext> options) : DbContext(options)
+{
+    public DbSet<SloTable> SloEntities { get; set; }
+
+    protected override void OnModelCreating(ModelBuilder modelBuilder) =>
+        modelBuilder.Entity<SloTable>()
+            .ToTable(SloTable.Name)
+            .HasKey(e => new { e.Guid, e.Id });
+}