diff --git a/src/NRedisStack/PublicAPI/PublicAPI.Unshipped.txt b/src/NRedisStack/PublicAPI/PublicAPI.Unshipped.txt index 37b6fe25..465a8ecf 100644 --- a/src/NRedisStack/PublicAPI/PublicAPI.Unshipped.txt +++ b/src/NRedisStack/PublicAPI/PublicAPI.Unshipped.txt @@ -43,3 +43,7 @@ NRedisStack.TimeSeriesCommandsAsync.MRangeAsync(NRedisStack.DataTypes.TimeStamp NRedisStack.TimeSeriesCommandsAsync.MRevRangeAsync(NRedisStack.DataTypes.TimeStamp fromTimeStamp, NRedisStack.DataTypes.TimeStamp toTimeStamp, System.Collections.Generic.IReadOnlyCollection! filter, bool latest = false, System.Collections.Generic.IReadOnlyCollection? filterByTs = null, (long, long)? filterByValue = null, bool? withLabels = null, System.Collections.Generic.IReadOnlyCollection? selectLabels = null, long? count = null, NRedisStack.DataTypes.TimeStamp? align = null, NRedisStack.TsAggregations aggregation = default(NRedisStack.TsAggregations), long? timeBucket = null, NRedisStack.Literals.Enums.TsBucketTimestamps? bt = null, bool empty = false, (string!, NRedisStack.Literals.Enums.TsReduce)? groupbyTuple = null) -> System.Threading.Tasks.Task! labels, System.Collections.Generic.IReadOnlyList! values)>!>! NRedisStack.TimeSeriesCommandsAsync.RangeAsync(string! key, NRedisStack.DataTypes.TimeStamp fromTimeStamp, NRedisStack.DataTypes.TimeStamp toTimeStamp, bool latest = false, System.Collections.Generic.IReadOnlyCollection? filterByTs = null, (long, long)? filterByValue = null, long? count = null, NRedisStack.DataTypes.TimeStamp? align = null, NRedisStack.TsAggregations aggregation = default(NRedisStack.TsAggregations), long? timeBucket = null, NRedisStack.Literals.Enums.TsBucketTimestamps? bt = null, bool empty = false) -> System.Threading.Tasks.Task!>! NRedisStack.TimeSeriesCommandsAsync.RevRangeAsync(string! key, NRedisStack.DataTypes.TimeStamp fromTimeStamp, NRedisStack.DataTypes.TimeStamp toTimeStamp, bool latest = false, System.Collections.Generic.IReadOnlyCollection? filterByTs = null, (long, long)? filterByValue = null, long? count = null, NRedisStack.DataTypes.TimeStamp? align = null, NRedisStack.TsAggregations aggregation = default(NRedisStack.TsAggregations), long? timeBucket = null, NRedisStack.Literals.Enums.TsBucketTimestamps? bt = null, bool empty = false) -> System.Threading.Tasks.Task!>! +NRedisStack.Search.SearchResult.Warnings.get -> string![]! +NRedisStack.Search.AggregationResult.Warnings.get -> string![]! +NRedisStack.Search.HybridSearchResult.Warnings.get -> string![]! +NRedisStack.Search.AggregationRequest.AddScores() -> NRedisStack.Search.AggregationRequest! diff --git a/src/NRedisStack/ResponseParser.cs b/src/NRedisStack/ResponseParser.cs index a0f781f1..e110730d 100644 --- a/src/NRedisStack/ResponseParser.cs +++ b/src/NRedisStack/ResponseParser.cs @@ -1289,6 +1289,34 @@ internal static T[] ParseSearchResultsMap(RedisResult result, Func + /// Exposes the full-text score values to the aggregation pipeline via the ADDSCORES option. + /// The score of each result is then available as the @__score field, e.g. for use in + /// SORTBY or APPLY. + /// + public AggregationRequest AddScores() + { + args.Add(SearchArgs.ADDSCORES); + return this; + } + public AggregationRequest Load(params FieldName[] fields) { if (fields.Length > 0) diff --git a/src/NRedisStack/Search/AggregationResult.cs b/src/NRedisStack/Search/AggregationResult.cs index 41420ed5..1a1fbcf7 100644 --- a/src/NRedisStack/Search/AggregationResult.cs +++ b/src/NRedisStack/Search/AggregationResult.cs @@ -27,6 +27,13 @@ internal WithCursorAggregationResult(string indexName, RedisResult result, long public long CursorId { get; } + /// + /// Warnings reported by the server for this query (for example when a query times out under the + /// return/return-strict on-timeout policy and partial results are returned). + /// Only populated on RESP3; on RESP2 FT.AGGREGATE does not carry warnings and this array is empty. + /// + public string[] Warnings { get; } = []; + internal AggregationResult(RedisResult result, long cursorId = -1) { var arr = (RedisResult[])result!; @@ -35,6 +42,7 @@ internal AggregationResult(RedisResult result, long cursorId = -1) { results = ResponseParser.ParseSearchResultsMap(result, ParseRecordFromMap, out long totalResults); // TotalResults = totalResults; // we ignore this in RESP3 mode for consistent API behaviour + Warnings = ResponseParser.ParseWarnings(result); } else { diff --git a/src/NRedisStack/Search/HybridSearchResult.cs b/src/NRedisStack/Search/HybridSearchResult.cs index f2dab59f..ed55423b 100644 --- a/src/NRedisStack/Search/HybridSearchResult.cs +++ b/src/NRedisStack/Search/HybridSearchResult.cs @@ -30,7 +30,6 @@ internal static HybridSearchResult Parse(RedisResult? result) case ResultKey.ExecutionTime: obj.ExecutionTime = TimeSpan.FromSeconds((double)value); break; - /* // defer Warnings until we've seen examples case ResultKey.Warnings when value.Length > 0: var warnings = new string[value.Length]; for (int j = 0; j < value.Length; j++) @@ -39,7 +38,6 @@ internal static HybridSearchResult Parse(RedisResult? result) } obj.Warnings = warnings; break; - */ case ResultKey.Results when value.Length > 0: obj._rawResults = value.ToArray(); break; @@ -118,8 +116,12 @@ private enum ResultKey /// public TimeSpan ExecutionTime { get; private set; } - // not exposing this until I've seen it being used - internal string[] Warnings { get; private set; } = []; + /// + /// Warnings reported by the server for this query (for example a timeout warning under the + /// return/return-strict on-timeout policy). FT.HYBRID reports warnings on both + /// RESP2 and RESP3. + /// + public string[] Warnings { get; private set; } = []; private RedisResult[] _rawResults = []; private Document[]? _docResults; diff --git a/src/NRedisStack/Search/Literals/CommandArgs.cs b/src/NRedisStack/Search/Literals/CommandArgs.cs index a139d65c..297d08d5 100644 --- a/src/NRedisStack/Search/Literals/CommandArgs.cs +++ b/src/NRedisStack/Search/Literals/CommandArgs.cs @@ -2,6 +2,7 @@ namespace NRedisStack.Search.Literals; internal class SearchArgs { + public const string ADDSCORES = "ADDSCORES"; public const string AGGREGATE = "AGGREGATE"; public const string APPLY = "APPLY"; public const string AS = "AS"; diff --git a/src/NRedisStack/Search/SearchResult.cs b/src/NRedisStack/Search/SearchResult.cs index 2294ee5e..93756e6a 100644 --- a/src/NRedisStack/Search/SearchResult.cs +++ b/src/NRedisStack/Search/SearchResult.cs @@ -13,6 +13,13 @@ public class SearchResult public long TotalResults { get; } public List Documents { get; } + /// + /// Warnings reported by the server for this query (for example when a query times out under the + /// return/return-strict on-timeout policy and partial results are returned). + /// Only populated on RESP3; on RESP2 FT.SEARCH does not carry warnings and this array is empty. + /// + public string[] Warnings { get; } = []; + /// /// Converts the documents to a list of json strings. only works on a json documents index. /// @@ -61,6 +68,7 @@ internal SearchResult(RedisResult root, bool hasContent, bool hasScores, bool ha return Document.Load(id, score, payload, fields, scoreExplained); }, out long totalResults)); TotalResults = totalResults; + Warnings = ResponseParser.ParseWarnings(root); } else // RESP2 { diff --git a/tests/NRedisStack.Tests/CustomAssertions.cs b/tests/NRedisStack.Tests/CustomAssertions.cs index c1a6f596..443ba97e 100644 --- a/tests/NRedisStack.Tests/CustomAssertions.cs +++ b/tests/NRedisStack.Tests/CustomAssertions.cs @@ -1,3 +1,4 @@ +using NRedisStack.Search; using Xunit; namespace NRedisStack.Tests; @@ -17,4 +18,38 @@ public static void LessThan(T actual, T expected) where T : IComparable Assert.True(actual.CompareTo(expected) < 0, $"Failure: Expected value to be less than {expected}, but found {actual}."); } + + /// + /// Asserts that RediSearch has indexed exactly documents in . + /// Indexing can lag behind the writes, so FT.INFO num_docs is re-read a few times to let indexing catch up before the + /// exact-equality assertion. + /// + public static void AssertIndexSize(ISearchCommands ft, string index, long expected) + { + long indexed = -1; + // allow search time to catch up + for (int i = 0; i < 10; i++) + { + indexed = ft.Info(index).NumDocs; + + if (indexed == expected) + break; + } + Assert.Equal(expected, indexed); + } + + /// + public static async Task AssertIndexSizeAsync(ISearchCommandsAsync ft, string index, long expected) + { + long indexed = -1; + // allow search time to catch up + for (int i = 0; i < 10; i++) + { + indexed = (await ft.InfoAsync(index)).NumDocs; + + if (indexed == expected) + break; + } + Assert.Equal(expected, indexed); + } } diff --git a/tests/NRedisStack.Tests/Search/HybridSearchIntegrationTests.cs b/tests/NRedisStack.Tests/Search/HybridSearchIntegrationTests.cs index b5581091..ae968e05 100644 --- a/tests/NRedisStack.Tests/Search/HybridSearchIntegrationTests.cs +++ b/tests/NRedisStack.Tests/Search/HybridSearchIntegrationTests.cs @@ -7,6 +7,8 @@ using NRedisStack.Search.Aggregation; using StackExchange.Redis; using Xunit; +using static NRedisStack.Tests.CustomAssertions; +using static NRedisStack.Tests.Search.SearchTestUtils; namespace NRedisStack.Tests.Search; @@ -432,4 +434,69 @@ static void WriteEscaped(ReadOnlySpan span, StringBuilder sb) sb.Append('"'); } } + + // Number of documents indexed by the on-timeout test. Enough that the hybrid query cannot + // complete within the 1ms per-query timeout, so the on-timeout policy is guaranteed to kick in. + private const int TimeoutDocCount = 10_000; + + // FT.HYBRID counterpart of the FT.SEARCH/FT.AGGREGATE on-timeout tests: under the `return` + // policy a timed-out query must return (possibly partial) results plus a warning rather than + // failing. FT.HYBRID reports warnings on both RESP2 and RESP3, so no protocol skip is needed. + // See CAE-3003. + [SkipIfRedisTheory(Comparison.LessThan, "8.9.0", deferVersionCheck: true)] + [MemberData(nameof(EndpointsFixture.Env.AllEnvironments), MemberType = typeof(EndpointsFixture.Env))] + public async Task TestHybridOnTimeoutReturnPopulatesWarnings(string endpointId) + { + var api = await CreateIndexAsync(endpointId, populate: false); + if (api.IsNull) return; +#if NET + // Bulk-load with fire-and-forget so many documents load quickly, then Ping so all writes + // have been processed before we wait for indexing. + var rand = new Random(12345); + using var vectorData = VectorData.Lease(V1DIM); + for (int i = 0; i < TimeoutDocCount; i++) + { + foreach (ref Half val in vectorData.Span) + { + val = (Half)rand.NextDouble(); + } + + HashEntry[] entry = + [ + new("text1", $"hello world {i}"), + new("vector1", vectorData.AsRedisValue()) + ]; + api.DB.HashSet($"{api.Index}_e{i}", entry, flags: CommandFlags.FireAndForget); + } + + api.DB.Ping(); + + AssertIndexSize(api.FT, api.Index, TimeoutDocCount); + + // Ensure the RETURN policy is active on every primary. A stored vector guarantees the query + // vector matches the field type/dimension. + SetSearchOnTimeout(api.DB.Multiplexer, "return"); + + var hash = (await api.DB.HashGetAllAsync($"{api.Index}_e0")).ToDictionary(k => k.Name, v => v.Value); + var vec = (byte[])hash["vector1"]!; + var query = new HybridSearchQuery() + .Search("hello world") + .VectorSearch(new("@vector1", VectorData.Raw(vec), + method: VectorSearchMethod.NearestNeighbour(TimeoutDocCount))) + .Combine(HybridSearchQuery.Combiner.Linear(0.5, 0.5, window: TimeoutDocCount)) + .ReturnFields("@text1") + .Apply(new("format(\"%s%s%s\",@text1,@text1,@text1)", "a"), new("format(\"%s%s%s\",@a,@a,@a)", "a"), + new("format(\"%s%s%s\",@a,@a,@a)", "a"), new("format(\"%s%s%s\",@a,@a,@a)", "a"), + new("format(\"%s%s%s\",@a,@a,@a)", "a")) + .SortBy(SortedField.Desc("@a")) + .Timeout(TimeSpan.FromMilliseconds(1)); + + var result = api.FT.HybridSearch(api.Index, query); + Assert.NotNull(result); + Assert.NotEmpty(result.Warnings); + Assert.Contains(result.Warnings, w => w.ToLowerInvariant().Contains("timeout")); +#else + await Task.CompletedTask; +#endif + } } \ No newline at end of file diff --git a/tests/NRedisStack.Tests/Search/SearchTestUtils.cs b/tests/NRedisStack.Tests/Search/SearchTestUtils.cs new file mode 100644 index 00000000..cab79758 --- /dev/null +++ b/tests/NRedisStack.Tests/Search/SearchTestUtils.cs @@ -0,0 +1,21 @@ +using StackExchange.Redis; + +namespace NRedisStack.Tests.Search; + +public static class SearchTestUtils +{ + // Sets the global search-on-timeout policy on every primary. On a cluster the policy must be + // present on all shards (and the coordinator) for the timeout behaviour to be consistent, so we + // broadcast rather than target a single node. + public static void SetSearchOnTimeout(IConnectionMultiplexer muxer, string value) + { + foreach (var endpoint in muxer.GetEndPoints()) + { + var server = muxer.GetServer(endpoint); + if (!server.IsReplica) + { + server.ConfigSet("search-on-timeout", value); + } + } + } +} diff --git a/tests/NRedisStack.Tests/Search/SearchTests.cs b/tests/NRedisStack.Tests/Search/SearchTests.cs index db02c559..855fa3f8 100644 --- a/tests/NRedisStack.Tests/Search/SearchTests.cs +++ b/tests/NRedisStack.Tests/Search/SearchTests.cs @@ -3,6 +3,8 @@ using StackExchange.Redis; using NRedisStack.RedisStackCommands; using NRedisStack.Search; +using static NRedisStack.Tests.CustomAssertions; +using static NRedisStack.Tests.Search.SearchTestUtils; using static NRedisStack.Search.Schema; using NRedisStack.Search.Aggregation; using NRedisStack.Search.Literals.Enums; @@ -70,34 +72,6 @@ private async Task AssertDatabaseSizeAsync(IDatabase db, int expected) Assert.Equal(expected, await DatabaseSizeAsync(db)); } - private void AssertIndexSize(ISearchCommands ft, string index, int expected) - { - long indexed = -1; - // allow search time to catch up - for (int i = 0; i < 10; i++) - { - indexed = ft.Info(index).NumDocs; - - if (indexed == expected) - break; - } - Assert.Equal(expected, indexed); - } - - private async Task AssertIndexSizeAsync(ISearchCommandsAsync ft, string index, int expected) - { - long indexed = -1; - // allow search time to catch up - for (int i = 0; i < 10; i++) - { - indexed = (await ft.InfoAsync(index)).NumDocs; - - if (indexed == expected) - break; - } - Assert.Equal(expected, indexed); - } - [Theory] [MemberData(nameof(EndpointsFixture.Env.AllEnvironments), MemberType = typeof(EndpointsFixture.Env))] public void TestSearchLimitOffset(string endpointId) @@ -4046,4 +4020,142 @@ public async Task TestDocumentLoadWithDB_Issue352(string endpointId) // Without fix for Issue352, document load in this case fails %100 with my local test runs,, and %100 success with fixed version. // The results in pipeline should be the same. } + + // Number of documents indexed by the on-timeout tests. Large enough that scanning, scoring and + // sorting them cannot complete within the 1ms per-query timeout, so the query engine's + // on-timeout policy is guaranteed to kick in. + private const int TimeoutDocCount = 10_000; + + private void PopulateTimeoutIndex(IDatabase db, SearchCommands ft) + { + Schema sc = new(); + sc.AddTextField("title"); + sc.AddNumericField("n", sortable: true); + ft.Create(index, FTCreateParams.CreateParams(), sc); + + // Bulk-load with fire-and-forget documents quickly, then Ping to make sure all + // writes have been processed before we wait for indexing. + for (int i = 0; i < TimeoutDocCount; i++) + { + db.HashSet($"tdoc:{i}", + new HashEntry[] { new("title", $"hello world {i}"), new("n", i) }, + flags: CommandFlags.FireAndForget); + } + + db.Ping(); + AssertIndexSize(ft, index, TimeoutDocCount); + } + + // A query heavy enough that it cannot finish within a 1ms timeout: it matches every document and + // forces a full scan, scoring and sort over the whole index. + private static Query TimingOutQuery() => + new Query("hello world").SetWithScores().SetSortBy("n", ascending: false).Limit(0, TimeoutDocCount).Timeout(1); + + // With the `fail` on-timeout policy the server returns an error on a timed-out search instead of + // partial results. The client must surface that error (as a RedisServerException) rather than + // swallowing it into an empty/partial result, and it must not be automatically retried. + // See CAE-3003 (initiative RED-132340: "Timeout guardrails"). + [SkipIfRedisTheory(Comparison.LessThan, "8.9.0")] + [MemberData(nameof(EndpointsFixture.Env.AllEnvironments), MemberType = typeof(EndpointsFixture.Env))] + public void TestSearchOnTimeoutFailReturnsError(string endpointId) + { + SkipClusterPre8(endpointId); + IDatabase db = GetCleanDatabase(endpointId); + var ft = db.FT(); + PopulateTimeoutIndex(db, ft); + + // Use the unified CONFIG SET (search-on-timeout); it is a global server setting. + SetSearchOnTimeout(db.Multiplexer, "fail"); + try + { + var ex = Assert.Throws(() => ft.Search(index, TimingOutQuery())); + Assert.Contains("timeout", ex.Message.ToLowerInvariant()); + } + finally + { + // Restore the default policy so we don't leak the FAIL setting into other tests. + SetSearchOnTimeout(db.Multiplexer, "return"); + } + } + + // With the `return` on-timeout policy the server returns a (possibly partial) result together + // with a warning instead of failing. The query must not throw, and the timeout must be reported + // via SearchResult.Warnings. Warnings are only carried on RESP3 for FT.SEARCH. See CAE-3003. + [SkipIfRedisTheory(Comparison.LessThan, "8.9.0")] + [MemberData(nameof(EndpointsFixture.Env.AllEnvironments), MemberType = typeof(EndpointsFixture.Env))] + public void TestSearchOnTimeoutReturnPopulatesWarnings(string endpointId) + { + SkipClusterPre8(endpointId); + Assert.SkipUnless(TestContext.Current.GetRunProtocol().IsResp3(), + "FT.SEARCH warnings are only returned on RESP3"); + + IDatabase db = GetCleanDatabase(endpointId); + var ft = db.FT(); + PopulateTimeoutIndex(db, ft); + + SetSearchOnTimeout(db.Multiplexer, "return"); + + var result = ft.Search(index, TimingOutQuery()); + Assert.NotNull(result); + Assert.NotEmpty(result.Warnings); + Assert.Contains(result.Warnings, w => w.ToLowerInvariant().Contains("timeout")); + } + + // A heavy FT.AGGREGATE that cannot finish within a 1ms timeout. A plain sort over the numeric + // field is optimized away (partial-range/skip-sorter), so instead we force the "no optimization" + // path documented for FT.AGGREGATE: score every document (ADDSCORES) and SORTBY @__score, load + // the text field (an HMGET per document) and blow it up with repeated string formatting. This + // costs tens of ms over ~1k documents, well beyond the 1ms budget on any hardware. + private static AggregationRequest TimingOutAggregation() => + new AggregationRequest("hello world").AddScores().Load(FieldName.Of("@title")) + .Apply("format(\"%s%s%s\",@title,@title,@title)", "a") + .Apply("format(\"%s%s%s\",@a,@a,@a)", "a") + .Apply("format(\"%s%s%s\",@a,@a,@a)", "a") + .Apply("format(\"%s%s%s\",@a,@a,@a)", "a") + .Apply("format(\"%s%s%s\",@a,@a,@a)", "a") + .SortBy(SortedField.Desc("@__score")).Timeout(1); + + // FT.AGGREGATE counterpart of TestSearchOnTimeoutFailReturnsError. See CAE-3003. + [SkipIfRedisTheory(Comparison.LessThan, "8.9.0")] + [MemberData(nameof(EndpointsFixture.Env.AllEnvironments), MemberType = typeof(EndpointsFixture.Env))] + public void TestAggregateOnTimeoutFailReturnsError(string endpointId) + { + SkipClusterPre8(endpointId); + IDatabase db = GetCleanDatabase(endpointId); + var ft = db.FT(); + PopulateTimeoutIndex(db, ft); + + SetSearchOnTimeout(db.Multiplexer, "fail"); + try + { + var ex = Assert.Throws(() => ft.Aggregate(index, TimingOutAggregation())); + Assert.Contains("timeout", ex.Message.ToLowerInvariant()); + } + finally + { + SetSearchOnTimeout(db.Multiplexer, "return"); + } + } + + // FT.AGGREGATE counterpart of TestSearchOnTimeoutReturnPopulatesWarnings. FT.AGGREGATE warnings + // are only carried on RESP3. See CAE-3003. + [SkipIfRedisTheory(Comparison.LessThan, "8.9.0")] + [MemberData(nameof(EndpointsFixture.Env.AllEnvironments), MemberType = typeof(EndpointsFixture.Env))] + public void TestAggregateOnTimeoutReturnPopulatesWarnings(string endpointId) + { + SkipClusterPre8(endpointId); + Assert.SkipUnless(TestContext.Current.GetRunProtocol().IsResp3(), + "FT.AGGREGATE warnings are only returned on RESP3"); + + IDatabase db = GetCleanDatabase(endpointId); + var ft = db.FT(); + PopulateTimeoutIndex(db, ft); + + SetSearchOnTimeout(db.Multiplexer, "return"); + + var result = ft.Aggregate(index, TimingOutAggregation()); + Assert.NotNull(result); + Assert.NotEmpty(result.Warnings); + Assert.Contains(result.Warnings, w => w.ToLowerInvariant().Contains("timeout")); + } } \ No newline at end of file