diff --git a/src/MeshWeaver.Documentation/Data/DataMesh/QuerySyntax.md b/src/MeshWeaver.Documentation/Data/DataMesh/QuerySyntax.md index 2ed9543c83..a3d0b94c67 100644 --- a/src/MeshWeaver.Documentation/Data/DataMesh/QuerySyntax.md +++ b/src/MeshWeaver.Documentation/Data/DataMesh/QuerySyntax.md @@ -531,6 +531,20 @@ difference is invisible at the call site: | **untyped** (`Query`, the MCP `search` tool) | each row becomes a `Dictionary` of exactly the named fields | | **typed on `MeshNode`** (`Query`, and therefore **every `workspace.GetQuery` / `hub.GetQuery`**) | the row stays a **`MeshNode`** — no dictionary — but the SQL projection is still narrowed | +**The untyped projection happens AFTER the merged sort** (#6830). An untyped read runs through +every provider and the cross-provider merge as nodes, and only the final, sorted and clipped +window is turned into dictionaries — so `select:path,name sort: limit:N` returns the +mesh's top N in the requested order, and the sort key need not be among the selected fields. +Before #6830 each provider projected its own rows first; the merge could not read a sort key off a +dictionary and clipped in provider-arrival order, so a query served by two providers (a store plus +the seeded static nodes, or several partitions) returned one provider's top N. + +A **free-text** read with no `sort:` is ordered by relevance, and the relevance travels with the +rows: the storage-adapter provider ranks its matches by fuzzy score before its load cap AND +reports that score on its Initial frame (`QueryResultChange.Scores`), so the merge re-sorts by the +same key (score, then path) instead of scoring every row 0 and answering path order. That held for +typed and untyped reads alike, and `Skip` pages the relevance order. + On the typed path only ONE column is actually conditional: **`content`**. Every other field is projected regardless, so naming them is documentation rather than optimisation. But when `content` is *not* named, the storage adapter emits `NULL::jsonb AS content` (Postgres) or omits the field diff --git a/src/MeshWeaver.Hosting/Persistence/Query/MeshQuery.cs b/src/MeshWeaver.Hosting/Persistence/Query/MeshQuery.cs index bd9514e76f..9794cc5a0c 100644 --- a/src/MeshWeaver.Hosting/Persistence/Query/MeshQuery.cs +++ b/src/MeshWeaver.Hosting/Persistence/Query/MeshQuery.cs @@ -237,6 +237,8 @@ internal static IReadOnlyList MergeQueryNamespaces(ParsedQuery parsed) /// An observable of merged query result changes across all matching providers. public IObservable> Query(MeshQueryRequest request) { + if (typeof(T) == typeof(object) && Projects(request)) + return (IObservable>)(object)ProjectAfterMerge(Query(request), request); var matched = SelectMatchingProviders(NamespacesForRequest(request)); // Run the subscribe on the DRAINABLE Query pool (not a bare TaskPoolScheduler): keeps the // calling hub's action block free while providers open their change feeds, AND makes the @@ -436,6 +438,9 @@ IObservable> IMeshQueryCore.Query( MeshQueryRequest request, JsonSerializerOptions options) { + if (typeof(T) == typeof(object) && Projects(request)) + return (IObservable>)(object)ProjectAfterMerge( + ((IMeshQueryCore)this).Query(request, options), request); var isSystem = string.IsNullOrEmpty(request.UserId) || string.Equals(request.UserId, WellKnownUsers.System, StringComparison.Ordinal); var matched = SelectMatchingProviders(NamespacesForRequest(request)); @@ -1102,6 +1107,53 @@ public IDisposable Arm( private static string PathTiebreak((T Item, double Score) hit) => hit.Item is MeshNode node ? node.Path ?? "" : ""; + /// Whether any of the request's queries carries a select: projection. + private static bool Projects(MeshQueryRequest request) + => request.EffectiveQueries.Any(q => new QueryParser().Parse(q).Select is { Count: > 0 }); + + /// + /// 🚨 An UNTYPED read (Query<object>, the select: projection surface) runs + /// through the providers and the merge AS NODES, and is projected here — AFTER the merged sort, + /// skip and clip (#6830). Only a read that carries a select: takes this route: without + /// one the providers hand back nodes already, and an untyped read keeps whatever its providers + /// answer. + /// + /// Each provider used to project its own rows to select: dictionaries before the + /// merge saw them. can only read a sort key, a path tiebreak + /// or a dedup key off a , so over dictionaries it fell back to score and + /// then provider ARRIVAL order: select:path,name sort:name-desc limit:3 over two + /// providers answered one provider's top three, not the mesh's — for every sort key, and for + /// the default newest-first ordering too. The sort key need not even be among the selected + /// fields, so the merge cannot be taught to read dictionaries; the projection has to wait. + /// The SQL providers still narrow their columns to the select: list (plus a content sort + /// key) when asked for nodes, so nothing extra is read. + /// + private static IObservable> ProjectAfterMerge( + IObservable> nodes, + MeshQueryRequest request) + => nodes.Select(change => + { + var parsed = change.Query + ?? new QueryParser().Parse(request.EffectiveQueries.FirstOrDefault() ?? ""); + var select = parsed.Select is { Count: > 0 } s ? s : null; + var items = change.Items + .Select(node => select is null ? (object)node : ParsedQuery.ProjectToSelect(node, select)) + .ToList(); + return new QueryResultChange + { + ChangeType = change.ChangeType, + Items = items, + Scores = change.Scores, + Partitions = change.Partitions, + PartiallyReadPartitions = change.PartiallyReadPartitions, + SilentProviders = change.SilentProviders, + SnapshotIncomplete = change.SnapshotIncomplete, + Query = change.Query!, + Version = change.Version, + Timestamp = change.Timestamp, + }; + }); + /// /// Sort + skip + clip the merged initial set. The authoritative ordering /// pass for every multi-provider Initial emission. Mirrors the post-collect diff --git a/src/MeshWeaver.Hosting/Persistence/Query/StorageAdapterMeshQueryProvider.cs b/src/MeshWeaver.Hosting/Persistence/Query/StorageAdapterMeshQueryProvider.cs index e04120ff14..d58cd9e4e8 100644 --- a/src/MeshWeaver.Hosting/Persistence/Query/StorageAdapterMeshQueryProvider.cs +++ b/src/MeshWeaver.Hosting/Persistence/Query/StorageAdapterMeshQueryProvider.cs @@ -206,6 +206,22 @@ private IObservable> RunQueryNodes( bool useSecurityFilter, bool projectSelect = false, ReadCompleteness? completeness = null) + => RunRankedQueryNodes(request, options, useSecurityFilter, projectSelect, completeness) + .Select(ranked => ranked.Rows); + + /// + /// plus the relevance it ranked by: Scores is one fuzzy + /// score per row, in row order, when a free-text term (and no sort:) made relevance the + /// ordering, and otherwise. The score is taken from the node as it was + /// MATCHED — before a projection or the content drop could change what the scorer reads — so + /// the merge re-sorts by exactly the key this provider clipped by (#6830 review). + /// + private IObservable<(IReadOnlyList Rows, IReadOnlyList? Scores)> RunRankedQueryNodes( + MeshQueryRequest request, + JsonSerializerOptions options, + bool useSecurityFilter, + bool projectSelect, + ReadCompleteness? completeness) => CollectMatched(request, options, useSecurityFilter, completeness) .Select(collected => { @@ -237,6 +253,9 @@ private IObservable> RunQueryNodes( // it has to be applied HERE, before the load cap — a merge-layer sort over rows this // provider already clipped in path order would order the wrong subset (#4950). IEnumerable sorted; + // Set only on the relevance branch: the score each row was ranked by, keyed by the + // matched node (reference identity), so it is read BEFORE projection. + Dictionary? relevance = null; var effectiveOrderBy = parsedQuery.EffectiveOrderBy; if (effectiveOrderBy != null) { @@ -263,8 +282,16 @@ private IObservable> RunQueryNodes( // OrderByDescending is stable — but "keeps the walk order" is exactly the // non-order above, so equal-score ties now break on path instead. // An explicit sort: (OrderBy above) still wins over relevance. + // The score is ALSO handed on (#6830 review): MeshQuery.ClipMergedInitial + // re-sorts every Initial by its Scores and then path, so ranking here without + // reporting the rank made every row 0 there and the merge answered PATH order. + var text = parsedQuery.TextSearch; + relevance = new Dictionary(ReferenceEqualityComparer.Instance); + foreach (var n in matchedNodes) + relevance[n] = _evaluator.GetFuzzyScore(n, text); + var scoreOf = relevance; sorted = matchedNodes - .OrderByDescending(n => _evaluator.GetFuzzyScore(n, parsedQuery.TextSearch)) + .OrderByDescending(n => scoreOf[n]) .ThenBy(PathTiebreak, StringComparer.Ordinal); } else @@ -291,20 +318,24 @@ private IObservable> RunQueryNodes( // // The row-level SELECT still narrows what the adapter loads, so a projected read costs // only what it asked for; the node simply stays a node for the caller that asked for nodes. - IEnumerable projected = sorted.Select(node => - projectSelect && parsedQuery.Select != null - ? ParsedQuery.ProjectToSelect(node, parsedQuery.Select) - : DropUnprojectedContent(node, parsedQuery)); - // Load cap = (Skip ?? 0) + Limit. request.Skip itself is applied // POST-MERGE in ClipMergedInitial (applying it here too would // double-skip); we only bound the load so a deep walk over a // 10 000-row subtree doesn't materialise everything for a 3-item page. var loadCap = LoadCap(request, parsedQuery); if (loadCap is int cap && cap > 0) - projected = projected.Take(cap); + sorted = sorted.Take(cap); + var kept = sorted.ToList(); - return (IReadOnlyList)projected.ToList(); + IReadOnlyList rows = kept + .Select(node => projectSelect && parsedQuery.Select != null + ? ParsedQuery.ProjectToSelect(node, parsedQuery.Select) + : DropUnprojectedContent(node, parsedQuery)) + .ToList(); + IReadOnlyList? scores = relevance is null + ? null + : kept.Select(node => relevance[node]).ToList(); + return (rows, scores); }); @@ -1609,17 +1640,23 @@ private IObservable> ObserveQueryInternal( // Per-subscription, never a field — see ReadCompleteness for why a field would mark // every other concurrent caller's snapshot incomplete and never clear. var completeness = new ReadCompleteness(); - IObservable> RunQuery() => - RunQueryNodes(request, options, useSecurityFilter, + IObservable<(List<(string? Path, T Item)> Rows, List? Scores)> RunQuery() => + RunRankedQueryNodes(request, options, useSecurityFilter, projectSelect: typeof(T) != typeof(MeshNode), completeness: completeness) - .Select(nodes => + .Select(ranked => { var results = new List<(string?, T)>(); - foreach (var item in nodes) - if (item is T typed) - results.Add((GetItemPath(item), typed)); - return results; + // Kept parallel to the rows that survive the type filter. + var scores = ranked.Scores is null ? null : new List(); + for (var i = 0; i < ranked.Rows.Count; i++) + { + if (ranked.Rows[i] is not T typed) + continue; + results.Add((GetItemPath(typed), typed)); + scores?.Add(ranked.Scores![i]); + } + return (results, scores); }); // Subscribe to persistence.Changes BEFORE running the initial query so @@ -1706,10 +1743,10 @@ void Route(DataChangeNotification n) disposables.Add( RunQuery().Subscribe( - initialResults => + initial => { var initialItems = new List(); - foreach (var (path, item) in initialResults) + foreach (var (path, item) in initial.Rows) { initialItems.Add(item); if (!string.IsNullOrEmpty(path)) @@ -1793,7 +1830,7 @@ void Route(DataChangeNotification n) changeBuffer .CoalesceWhileRunning(RunQuery) .Subscribe( - newResults => ProcessBatch(newResults, currentItems, parsedQuery, observer), + newResults => ProcessBatch(newResults.Rows, currentItems, parsedQuery, observer), ex => observer.OnError(ex))); // Publish the live buffer and flip the gate in ONE critical section, so @@ -1811,6 +1848,13 @@ void Route(DataChangeNotification n) { ChangeType = QueryChangeType.Initial, Items = initialItems, + // 🚨 The relevance this provider RANKED by travels with the rows + // (#6830 review). RunQueryNodes orders a free-text read by fuzzy score + // before its load cap, but the merge (MeshQuery.ClipMergedInitial) + // re-sorts by Scores and then path — so a frame without them made every + // row score 0 and handed back PATH order, weaker matches first and Skip + // paging the wrong window. + Scores = initial.Scores, Query = parsedQuery, Version = Interlocked.Increment(ref _version), Timestamp = DateTimeOffset.UtcNow, diff --git a/test/MeshWeaver.Hosting.Test/UntypedProjectionMergedSortTest.cs b/test/MeshWeaver.Hosting.Test/UntypedProjectionMergedSortTest.cs new file mode 100644 index 0000000000..c407b0f49c --- /dev/null +++ b/test/MeshWeaver.Hosting.Test/UntypedProjectionMergedSortTest.cs @@ -0,0 +1,249 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Reactive.Linq; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; +using MeshWeaver.Fixture; +using MeshWeaver.Hosting.Persistence; +using MeshWeaver.Hosting.Persistence.Query; +using MeshWeaver.Mesh; +using MeshWeaver.Mesh.Services; +using Xunit; + +namespace MeshWeaver.Hosting.Test; + +/// +/// 🚨 An UNTYPED projection (select: read as Query<object>) is clipped in the +/// order the author asked for when two providers contribute (#6830). +/// +/// Each provider used to project its rows to select: dictionaries itself, so the +/// merge received rows it could not read a sort key from: ClipMergedInitial runs the +/// authoritative sort only over MeshNodes and fell back to score + path — and a dictionary +/// has no path, so the window was taken in provider-ARRIVAL order. select:path,name +/// sort:name-desc limit:3 over two providers then answered one provider's top three, not the +/// mesh's. The fix runs an untyped query as nodes through the merge and projects AFTER the merged +/// sort and clip. +/// +/// The fixture interleaves the two providers so every wrong answer is visible: the names are +/// A–F, the even letters in the storage-backed provider and the odd in the static-node provider, so the global top three by +/// name-desc (F, E, D) takes rows from BOTH, and either provider's own top three is +/// wrong. +/// +public class UntypedProjectionMergedSortTest +{ + private static readonly JsonSerializerOptions Options = new(); + + private static MeshNode Node(string id, string name) => new(id, "Book") + { + Name = name, + NodeType = "Chapter", + State = MeshNodeState.Active, + }; + + /// + /// The pairing every portal runs: a storage-backed provider and the static-node provider + /// (seeded nodes, built-in catalogues). The two must carry different provider names — the + /// fan-out folds providers by name. + /// + private static async Task TwoProviders(CancellationToken cancellationToken) + { + var store = new InMemoryStorageAdapter(); + // Path order disagrees with name order inside each provider too, so a path-ordered clip is + // wrong as well. + foreach (var (id, name) in new[] { ("a1", "F"), ("a2", "D"), ("a3", "B") }) + await store.Write(Node(id, name), Options).Await(cancellationToken); + var seeded = new StaticMeshNodeListProvider([Node("b1", "E"), Node("b2", "C"), Node("b3", "A")]); + return new MeshQuery( + [new StorageAdapterMeshQueryProvider(persistence: store), new StaticNodeQueryProvider([seeded], _ => true)], + hub: null!); + } + + private static async Task> Items( + IMeshQueryCore query, string text, CancellationToken cancellationToken) + { + var change = await query.Query(new MeshQueryRequest { Query = text }, Options) + .Where(c => c.ChangeType == QueryChangeType.Initial) + .FirstAsync() + .Timeout(TestTimeouts.Convergence) + .Await(cancellationToken); + return change.Items; + } + + private static string? Field(object row, string key) => + row is IReadOnlyDictionary dict && dict.TryGetValue(key, out var value) + ? value?.ToString() + : throw new InvalidOperationException($"an untyped select: row is a projection, got {row.GetType().Name}"); + + [Theory(Timeout = 60_000)] + [InlineData("namespace:Book nodeType:Chapter select:path,name sort:name-desc limit:3", "F", "E", "D")] + [InlineData("namespace:Book nodeType:Chapter select:path,name sort:name limit:3", "A", "B", "C")] + [InlineData("namespace:Book nodeType:Chapter select:name sort:path-desc limit:2", "A", "C")] + public async Task AnUntypedProjectionIsClippedInTheRequestedOrder(string text, params string[] expected) + { + var query = await TwoProviders(TestContext.Current.CancellationToken); + + var rows = await Items(query, text, TestContext.Current.CancellationToken); + + rows.Select(r => Field(r, "name")).Should().Equal(expected, + $"'{text}' over two providers answers the mesh's window in the requested order, not one provider's"); + } + + /// + /// The projection is still a projection: an untyped select: row carries exactly the + /// selected fields, so deferring it past the sort did not hand untyped callers whole nodes. + /// + [Fact(Timeout = 60_000)] + public async Task TheRowsAreStillProjections() + { + var query = await TwoProviders(TestContext.Current.CancellationToken); + + var rows = await Items(query, "namespace:Book nodeType:Chapter select:path,name sort:name-desc limit:3", + TestContext.Current.CancellationToken); + + rows.Should().OnlyContain(r => r is IReadOnlyDictionary, "an untyped select: row is a projection"); + foreach (var row in rows.Cast>()) + row.Keys.Select(k => k.ToLowerInvariant()).OrderBy(k => k, StringComparer.Ordinal) + .Should().Equal(new[] { "name", "path" }, "exactly the selected fields"); + rows.Select(r => Field(r, "path")).Should().Equal("Book/a1", "Book/b1", "Book/a2"); + } + + /// + /// The control: the TYPED read of the same query was already right (#6704) and must stay so — + /// the untyped answer now agrees with it row for row. + /// + [Fact(Timeout = 60_000)] + public async Task TheUntypedAnswerAgreesWithTheTypedOne() + { + var query = await TwoProviders(TestContext.Current.CancellationToken); + const string text = "namespace:Book nodeType:Chapter select:path,name sort:name-desc limit:3"; + + var typed = await Items(query, text, TestContext.Current.CancellationToken); + var untyped = await Items(query, text, TestContext.Current.CancellationToken); + + untyped.Select(r => Field(r, "path")).Should().Equal(typed.Select(n => n.Path)); + } + + // ── free text: the provider's relevance survives the merge (#6830 review) ───────────────── + + /// + /// Matches for needle whose relevance order is the REVERSE of their path order: the + /// exact name sits on the last path, the weakest (longest, term buried) on the first. + /// + private static readonly (string Id, string Name)[] Needles = + [ + ("n1", "a long haystack title that hides the word needle near its end"), + ("n2", "the needle somewhere"), + ("n3", "needle"), + ]; + + private static async Task OneStorageProvider(CancellationToken cancellationToken) + { + var store = new InMemoryStorageAdapter(); + foreach (var (id, name) in Needles) + await store.Write(Node(id, name), Options).Await(cancellationToken); + return new MeshQuery([new StorageAdapterMeshQueryProvider(persistence: store)], hub: null!); + } + + /// + /// The relevance order, from the very scorer the provider ranks by — never hand-guessed — and + /// asserted to DISAGREE with path order, so the test cannot pass by a path-ordered answer. + /// + private static string[] RelevanceOrder() + { + var evaluator = new QueryEvaluator(); + var ranked = Needles + .Select(n => (n.Name, Path: $"Book/{n.Id}", Score: evaluator.GetFuzzyScore(Node(n.Id, n.Name), "needle"))) + .OrderByDescending(r => r.Score) + .ThenBy(r => r.Path, StringComparer.Ordinal) + .Select(r => r.Name) + .ToArray(); + ranked.SequenceEqual(Needles.Select(n => n.Name)).Should().BeFalse( + "the fixture is built so relevance order and path order disagree"); + return ranked; + } + + public static TheoryData FreeTextQueries => new() + { + "namespace:Book needle select:path,name limit:3", + "namespace:Book needle limit:3", + }; + + /// + /// The core entry point (), typed and untyped, with and without a + /// select:: a free-text read with no sort: answers best match first. + /// + [Theory(Timeout = 60_000)] + [MemberData(nameof(FreeTextQueries))] + public async Task AFreeTextReadKeepsTheProvidersRelevanceOrder_Core(string text) + { + var ct = TestContext.Current.CancellationToken; + var query = (IMeshQueryCore)await OneStorageProvider(ct); + var expected = RelevanceOrder(); + + (await Items(query, text, ct)).Select(n => n.Name).Should().Equal(expected, + "a typed free-text read is ranked by relevance, not by path"); + (await Items(query, text, ct)).Select(r => r is MeshNode n ? n.Name : Field(r, "name")) + .Should().Equal(expected, "an untyped free-text read is ranked by relevance, not by path"); + } + + /// + /// The public entry point () merges the SECURED + /// provider surface () through the same + /// ClipMergedInitial as the core entry — and resolves its serializer options from a hub, + /// so it is pinned here at the frame it consumes: that surface's Initial carries one score per + /// row, in relevance order, for typed and untyped reads alike. Without them the merge scores + /// every row 0 and answers path order. + /// + [Theory(Timeout = 60_000)] + [MemberData(nameof(FreeTextQueries))] + public async Task AFreeTextReadKeepsTheProvidersRelevanceOrder_Public(string text) + { + var ct = TestContext.Current.CancellationToken; + var store = new InMemoryStorageAdapter(); + foreach (var (id, name) in Needles) + await store.Write(Node(id, name), Options).Await(ct); + IMeshQueryProvider provider = new StorageAdapterMeshQueryProvider(persistence: store); + var expected = RelevanceOrder(); + + async Task> Initial() => + await provider.Query(new MeshQueryRequest { Query = text }, Options) + .Where(c => c.ChangeType == QueryChangeType.Initial) + .FirstAsync() + .Timeout(TestTimeouts.Convergence) + .Await(ct); + + var typed = await Initial(); + typed.Items.Select(n => n.Name).Should().Equal(expected, "the provider ranks by relevance"); + typed.Scores.Should().NotBeNull("the rank the provider clipped by must reach the merge"); + typed.Scores!.Should().HaveCount(typed.Items.Count).And.BeInDescendingOrder(); + typed.Scores!.Distinct().Should().HaveCount(typed.Scores!.Count, + "the fixture's matches differ in relevance, so path order cannot satisfy the merge"); + + var untyped = await Initial(); + untyped.Scores.Should().NotBeNull("an untyped read reports its rank too"); + untyped.Scores!.Should().Equal(typed.Scores, "the score is read off the matched node, before any projection"); + } + + /// + /// Skip pages the RELEVANCE order: the second-best match is page two of size one. + /// + [Fact(Timeout = 60_000)] + public async Task SkipPagesTheRelevanceOrder() + { + var ct = TestContext.Current.CancellationToken; + var query = (IMeshQueryCore)await OneStorageProvider(ct); + var expected = RelevanceOrder(); + + var change = await query.Query( + new MeshQueryRequest { Query = "namespace:Book needle select:path,name", Skip = 1, Limit = 1 }, Options) + .Where(c => c.ChangeType == QueryChangeType.Initial) + .FirstAsync() + .Timeout(TestTimeouts.Convergence) + .Await(ct); + + change.Items.Select(r => Field(r, "name")).Should().Equal(new[] { expected[1] }, + "page two of the relevance order, not of the path order"); + } +}