Added stream index implementation

Author: oskardudycz

src/KurrentDB.SecondaryIndexing.Tests/IntegrationTests/Fixtures/SecondaryIndexingFixture.cs

@@ -56,7 +56,7 @@ public async Task<List<ResolvedEvent>> ReadUntil(
 		TimeSpan? timeout = null,
 		CancellationToken ct = default
 	) {
-		timeout ??= TimeSpan.FromMilliseconds(5000);
+		timeout ??= TimeSpan.FromMilliseconds(10000);
 		var endTime = DateTime.UtcNow.Add(timeout.Value);
 
 		var events = new List<ResolvedEvent>();
@@ -68,7 +68,7 @@ public async Task<List<ResolvedEvent>> ReadUntil(
 			try {
 				events = await ReadStream(streamName, ct).ToListAsync(ct);
 
-				reachedPosition = events.Count != 0 && events.Last().Event.LogPosition <= (long)position.CommitPosition;
+				reachedPosition = events.Count != 0 && events.Last().Event.LogPosition >= (long)position.CommitPosition;
 			} catch (ReadResponseException.StreamNotFound ex) {
 				streamNotFound = ex;
 			}

src/KurrentDB.SecondaryIndexing.Tests/IntegrationTests/SecondaryIndexingPluginEnabledIntegrationTests.cs

@@ -56,8 +56,8 @@ public async Task ReadsIndexStream_ForEnabledPlugin() {
 		Assert.NotEmpty(eventTypeReadResult);
 
 		var allResults = allReadResult.Where(e => e.Event.EventStreamId == streamName).ToList();
-		var categoryResults = allReadResult.Where(e => e.Event.EventStreamId == streamName).ToList();
-		var eventTypeResults = allReadResult.Where(e => e.Event.EventStreamId == streamName).ToList();
+		var categoryResults = categoryReadResult.Where(e => e.Event.EventStreamId == streamName).ToList();
+		var eventTypeResults = eventTypeReadResult.Where(e => e.Event.EventStreamId == streamName).ToList();
 
 		Assert.Equal(_expectedEventData.Length, allResults.Count);
 		Assert.Equal(_expectedEventData.Length, categoryResults.Count);

src/KurrentDB.SecondaryIndexing/Indices/Default/DefaultIndex.cs

@@ -8,6 +8,7 @@
 using KurrentDB.Core.Services.Storage.ReaderIndex;
 using KurrentDB.SecondaryIndexing.Indices.Category;
 using KurrentDB.SecondaryIndexing.Indices.EventType;
+using KurrentDB.SecondaryIndexing.Indices.Stream;
 using KurrentDB.SecondaryIndexing.Storage;
 
 namespace KurrentDB.SecondaryIndexing.Indices.Default;
@@ -25,6 +26,10 @@ internal class DefaultIndex<TStreamId> : Disposable, ISecondaryIndex {
 
 	public CategoryIndex<TStreamId> CategoryIndex { get; }
 
+	public EventTypeIndex<TStreamId> EventTypeIndex { get; set; }
+
+	public StreamIndex StreamIndex { get; set; }
+
 	public DefaultIndex(DuckDbDataSource db, IReadIndex<TStreamId> readIndex) {
 		_db = db;
 		_db.InitDb();
@@ -33,6 +38,7 @@ public DefaultIndex(DuckDbDataSource db, IReadIndex<TStreamId> readIndex) {
 
 		CategoryIndex = new CategoryIndex<TStreamId>(db, connection, readIndex);
 		EventTypeIndex = new EventTypeIndex<TStreamId>(db, connection, readIndex);
+		StreamIndex = new StreamIndex(connection);
 
 		var processor = new DefaultSecondaryIndexProcessor<TStreamId>(connection, this);
 		Processor = processor;
@@ -43,8 +49,6 @@ public DefaultIndex(DuckDbDataSource db, IReadIndex<TStreamId> readIndex) {
 		];
 	}
 
-	public EventTypeIndex<TStreamId> EventTypeIndex { get; set; }
-
 	public void Init() {
 	}
 

src/KurrentDB.SecondaryIndexing/Indices/Default/DefaultSecondaryIndexProcessor.cs

@@ -40,16 +40,18 @@ public async ValueTask Index(ResolvedEvent resolvedEvent, CancellationToken toke
 
 		await _defaultIndex.CategoryIndex.Processor.Index(resolvedEvent, token);
 		await _defaultIndex.EventTypeIndex.Processor.Index(resolvedEvent, token);
+		await _defaultIndex.StreamIndex.Processor.Index(resolvedEvent, token);
 
 		var category = _defaultIndex.CategoryIndex.LastIndexed;
 		var eventType = _defaultIndex.EventTypeIndex.LastIndexed;
+		var streamId = _defaultIndex.StreamIndex.LastIndexed;
 
 		using (var row = _appender.CreateRow()) {
 			row.Append(_seq++);
 			row.Append((int)resolvedEvent.Event.EventNumber);
 			row.Append(resolvedEvent.Event.LogPosition);
 			row.Append(resolvedEvent.Event.TimeStamp);
-			row.Append(0);
+			row.Append(streamId);
 			row.Append((int)eventType.Id);
 			row.Append(eventType.Sequence);
 			row.Append((int)category.Id);
@@ -71,5 +73,6 @@ public async ValueTask Commit(CancellationToken token = default) {
 
 		await _defaultIndex.CategoryIndex.Processor.Commit(token);
 		await _defaultIndex.EventTypeIndex.Processor.Commit(token);
+		await _defaultIndex.StreamIndex.Processor.Commit(token);
 	}
 }

src/KurrentDB.SecondaryIndexing/Indices/Stream/StreamIndex.cs

@@ -0,0 +1,26 @@
+// Copyright (c) Event Store Ltd and/or licensed to Event Store Ltd under one or more agreements.
+// Event Store Ltd licenses this file to you under the Event Store License v2 (see LICENSE.md).
+
+using Kurrent.Quack;
+using KurrentDB.Core.Services.Storage.InMemory;
+
+namespace KurrentDB.SecondaryIndexing.Indices.Stream;
+
+internal class StreamIndex(DuckDBAdvancedConnection connection) : ISecondaryIndex {
+	private readonly StreamIndexProcessor _processor = new(connection);
+
+	public void Init() {
+	}
+
+	public ulong? GetLastPosition() =>
+		(ulong)_processor.LastCommittedPosition;
+
+	public ulong? GetLastSequence() => (ulong)_processor.Seq;
+
+	public ISecondaryIndexProcessor Processor => _processor;
+	public IReadOnlyList<IVirtualStreamReader> Readers { get; } = [];
+
+	public long LastIndexed => _processor.LastIndexed;
+
+	public void Dispose() { }
+}

src/KurrentDB.SecondaryIndexing/Indices/Stream/StreamIndexProcessor.cs

@@ -0,0 +1,65 @@
+// Copyright (c) Kurrent, Inc and/or licensed to Kurrent, Inc under one or more agreements.
+// Kurrent, Inc licenses this file to you under the Kurrent License v1 (see LICENSE.md).
+
+using DotNext;
+using Kurrent.Quack;
+using KurrentDB.Core.Data;
+using KurrentDB.SecondaryIndexing.Storage;
+using Microsoft.Extensions.Caching.Memory;
+using static KurrentDB.SecondaryIndexing.Indices.Stream.StreamSql;
+
+namespace KurrentDB.SecondaryIndexing.Indices.Stream;
+
+internal class StreamIndexProcessor(DuckDBAdvancedConnection connection) : Disposable, ISecondaryIndexProcessor {
+	private readonly MemoryCache _streamIdCache = new(new MemoryCacheOptions());
+	private readonly MemoryCacheEntryOptions _options = new() { SlidingExpiration = TimeSpan.FromMinutes(10) };
+	private long _lastLogPosition;
+	private readonly Appender _appender = new(connection, "streams"u8);
+	public long Seq { get; private set; } = connection.QueryFirstOrDefault<long, QueryStreamsMaxSequencesSql>() ?? 0;
+	public long LastCommittedPosition { get; private set; }
+	public long LastIndexed { get; private set; }
+
+	public ValueTask Index(ResolvedEvent resolvedEvent, CancellationToken token = default) {
+		if (IsDisposingOrDisposed)
+			return ValueTask.CompletedTask;
+
+		var name = resolvedEvent.OriginalStreamId;
+		_lastLogPosition = resolvedEvent.Event.LogPosition;
+
+		if (_streamIdCache.TryGetValue(name, out var existing)) {
+			LastIndexed = (long)existing!;
+			return ValueTask.CompletedTask;
+		}
+
+		var fromDb = connection.QueryFirstOrDefault<QueryStreamArgs, long, QueryStreamIdSql>(
+			new QueryStreamArgs { StreamName = name }
+		);
+		if (fromDb.HasValue) {
+			_streamIdCache.Set(name, fromDb, _options);
+			LastIndexed = fromDb.Value;
+			return ValueTask.CompletedTask;
+		}
+
+		var id = ++Seq;
+		_streamIdCache.Set(name, id, _options);
+
+		using var row = _appender.CreateRow();
+		row.Append(id);
+		row.Append(name);
+		row.AppendDefault();
+		row.AppendDefault();
+
+		LastIndexed = id;
+		return ValueTask.CompletedTask;
+	}
+
+	public ValueTask Commit(CancellationToken token = default) {
+		if (IsDisposingOrDisposed)
+			return ValueTask.CompletedTask;
+
+		_appender.Flush();
+		LastCommittedPosition = _lastLogPosition;
+
+		return ValueTask.CompletedTask;
+	}
+}

src/KurrentDB.SecondaryIndexing/Indices/Stream/StreamSql.cs

@@ -0,0 +1,32 @@
+// Copyright (c) Kurrent, Inc and/or licensed to Kurrent, Inc under one or more agreements.
+// Kurrent, Inc licenses this file to you under the Kurrent License v1 (see LICENSE.md).
+
+using Kurrent.Quack;
+
+namespace KurrentDB.SecondaryIndexing.Indices.Stream;
+
+internal static class StreamSql {
+	public struct QueryStreamArgs {
+		public string StreamName { get; set; }
+	}
+
+	public struct QueryStreamIdSql : IQuery<QueryStreamArgs, long> {
+		public static BindingContext Bind(in QueryStreamArgs args, PreparedStatement statement) => new(statement) {
+			args.StreamName
+		};
+
+		public static ReadOnlySpan<byte> CommandText =>
+			"select id from streams where name=$1"u8;
+
+		public static long Parse(ref DataChunk.Row row) =>
+			row.ReadInt64();
+	}
+
+	public struct QueryStreamsMaxSequencesSql : IQuery<long> {
+		public static ReadOnlySpan<byte> CommandText =>
+			"select max(id) from streams"u8;
+
+		public static long Parse(ref DataChunk.Row row) =>
+			row.ReadInt64();
+	}
+}

src/KurrentDB.SecondaryIndexing/Storage/1_Schema.sql

@@ -1,42 +1,42 @@
 create table if not exists event_type (
-    id int4 primary key,
-    name varchar,
-    unique(name)
+										  id int4 primary key,
+										  name varchar not null,
+										  unique(name)
 );
 
 create table if not exists category (
-    id int4 primary key,
-    name varchar,
-    unique(name)
+										id int4 primary key,
+										name varchar not null,
+										unique(name)
 );
 
 create table if not exists streams (
-    id ubigint primary key,
-    name varchar,
-    unique(name),
-    max_age int4,
-    max_count int4
+									   id ubigint primary key,
+									   name varchar not null,
+									   unique(name),
+									   max_age int4 DEFAULT NULL,
+									   max_count int4 DEFAULT NULL
 );
 
 create table if not exists idx_all (
-    seq ubigint,
-    event_number int4,
-    log_position ubigint,
-    created timestamp,
-    stream ubigint,
-    event_type int4,
-    event_type_seq int8,
-    category int4,
-    category_seq int8
+									   seq ubigint not null,
+									   event_number int4 not null,
+									   log_position ubigint not null,
+									   created timestamp not null,
+									   stream ubigint not null,
+									   event_type int4 not null,
+									   event_type_seq int8 not null,
+									   category int4 not null,
+									   category_seq int8 not null
 );
 
 create index if not exists idx_all_category on idx_all(category, category_seq);
 create index if not exists idx_all_event_type on idx_all(event_type, category_seq);
 create index if not exists idx_sequence on idx_all(seq);
 
 create or replace macro read_category(name, startAt, finishAt) as table
-select
-	category_seq as seq,
+																	  select
+																  category_seq as seq,
 	event->>'stream_id' as stream_id,
 	event_number,
 	event->>'event_type' as event_type,
@@ -53,8 +53,8 @@ from (
 ) order by category_seq;
 
 create or replace macro read_all(position) as table
-select
-	seq,
+												  select
+											  seq,
 	event->>'stream_id' as stream_id,
 	event_number,
 	event->>'event_type' as event_type,
@@ -67,8 +67,8 @@ from (
 );
 
 create or replace macro read_category(name, start, count) as table
-select
-	category_seq as seq,
+																 select
+															 category_seq as seq,
 	event->>'stream_id' as stream_id,
 	event_number,
 	event->>'event_type' as event_type,

src/KurrentDB.SecondaryIndexing/Storage/DuckDbExtensions.cs

@@ -11,23 +11,36 @@ public static class DuckDbExtensions {
 		where TRow : struct
 		where TQuery : IParameterlessStatement, IDataRowParser<TRow> {
 		using (pool.Rent(out var connection)) {
-			using var result = connection.ExecuteQuery<TRow, TQuery>().GetEnumerator();
-
-			return result.MoveNext() ? null : result.Current;
+			return connection.QueryFirstOrDefault<TRow, TQuery>();
 		}
 	}
 
+	public static TRow? QueryFirstOrDefault<TRow, TQuery>(this DuckDBAdvancedConnection connection)
+		where TRow : struct
+		where TQuery : IParameterlessStatement, IDataRowParser<TRow> {
+		using var result = connection.ExecuteQuery<TRow, TQuery>().GetEnumerator();
+
+		return result.MoveNext() ? null : result.Current;
+	}
+
 	public static TRow? QueryFirstOrDefault<TArgs, TRow, TQuery>(this DuckDBConnectionPool pool, TArgs args)
 		where TArgs : struct
 		where TRow : struct
 		where TQuery : IPreparedStatement<TArgs>, IDataRowParser<TRow> {
 		using (pool.Rent(out var connection)) {
-			using var result = connection.ExecuteQuery<TArgs, TRow, TQuery>(args).GetEnumerator();
-
-			return result.MoveNext() ? null : result.Current;
+			return connection.QueryFirstOrDefault<TArgs, TRow, TQuery>(args);
 		}
 	}
 
+	public static TRow? QueryFirstOrDefault<TArgs, TRow, TQuery>(this DuckDBAdvancedConnection connection, TArgs args)
+		where TArgs : struct
+		where TRow : struct
+		where TQuery : IPreparedStatement<TArgs>, IDataRowParser<TRow> {
+		using var result = connection.ExecuteQuery<TArgs, TRow, TQuery>(args).GetEnumerator();
+
+		return result.MoveNext() ? null : result.Current;
+	}
+
 	public static List<TRow> Query<TRow, TQuery>(this DuckDBConnectionPool pool)
 		where TRow : struct
 		where TQuery : IDataRowParser<TRow>, IParameterlessStatement {