Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
14 changes: 14 additions & 0 deletions src/MeshWeaver.Documentation/Data/DataMesh/QuerySyntax.md
Original file line number Diff line number Diff line change
Expand Up @@ -531,6 +531,20 @@ difference is invisible at the call site:
| **untyped** (`Query<object>`, the MCP `search` tool) | each row becomes a `Dictionary<string, object>` of exactly the named fields |
| **typed on `MeshNode`** (`Query<MeshNode>`, 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:<anything> 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
Expand Down
52 changes: 52 additions & 0 deletions src/MeshWeaver.Hosting/Persistence/Query/MeshQuery.cs
Original file line number Diff line number Diff line change
Expand Up @@ -237,6 +237,8 @@ internal static IReadOnlyList<string> MergeQueryNamespaces(ParsedQuery parsed)
/// <returns>An observable of merged query result changes across all matching providers.</returns>
public IObservable<QueryResultChange<T>> Query<T>(MeshQueryRequest request)
{
if (typeof(T) == typeof(object) && Projects(request))
return (IObservable<QueryResultChange<T>>)(object)ProjectAfterMerge(Query<MeshNode>(request), request);
Comment on lines +240 to +241

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Confirmed and fixed in 6bda942. The provider ranked free-text matches by fuzzy score before its load cap, but it left Scores null, so ClipMergedInitial scored every row 0 and returned path order. That was already true for TYPED free-text reads, and routing untyped select: reads through the node merge extended it to them. The root fix is in StorageAdapterMeshQueryProvider. RunRankedQueryNodes takes the score off the matched node before any projection or content drop and carries it on the Initial frame, aligned with the rows that pass the type filter. Live deltas are unchanged.

Regressions in UntypedProjectionMergedSortTest use a fixture where relevance order is the reverse of path order. The expected order comes from QueryEvaluator.GetFuzzyScore itself, with a guard that it disagrees with path order.

  • Core entry (IMeshQueryCore.Query), typed and untyped, with and without select:.
  • Skip=1 Limit=1 pages the relevance order.
  • The public entry (MeshQuery.Query<T>(request)) needs a hub for its serializer options. It is pinned at the frame it merges: the secured IMeshQueryProvider.Query<T> Initial carries one descending score per row, the same for typed and untyped reads.

Negative control: with Scores = null restored, 5 of the 10 cases fail (for example, expected needle, the needle somewhere, … but got the path order a long haystack…, the needle somewhere, needle). With the fix all 10 pass, and so do the adjacent query suites (PagedQueryTotalOrder, MeshQueryMergeContract, ContentFieldSort, FilterOnlyQueryClipsNewestFirst).

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
Expand Down Expand Up @@ -436,6 +438,9 @@ IObservable<QueryResultChange<T>> IMeshQueryCore.Query<T>(
MeshQueryRequest request,
JsonSerializerOptions options)
{
if (typeof(T) == typeof(object) && Projects(request))
return (IObservable<QueryResultChange<T>>)(object)ProjectAfterMerge(
((IMeshQueryCore)this).Query<MeshNode>(request, options), request);
var isSystem = string.IsNullOrEmpty(request.UserId)
|| string.Equals(request.UserId, WellKnownUsers.System, StringComparison.Ordinal);
var matched = SelectMatchingProviders(NamespacesForRequest(request));
Expand Down Expand Up @@ -1102,6 +1107,53 @@ public IDisposable Arm(
private static string PathTiebreak<T>((T Item, double Score) hit)
=> hit.Item is MeshNode node ? node.Path ?? "" : "";

/// <summary>Whether any of the request's queries carries a <c>select:</c> projection.</summary>
private static bool Projects(MeshQueryRequest request)
=> request.EffectiveQueries.Any(q => new QueryParser().Parse(q).Select is { Count: > 0 });

/// <summary>
/// 🚨 An UNTYPED read (<c>Query&lt;object&gt;</c>, the <c>select:</c> 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 <c>select:</c> takes this route: without
/// one the providers hand back nodes already, and an untyped read keeps whatever its providers
/// answer.
///
/// <para>Each provider used to project its own rows to <c>select:</c> dictionaries before the
/// merge saw them. <see cref="ClipMergedInitial{T}"/> can only read a sort key, a path tiebreak
/// or a dedup key off a <see cref="MeshNode"/>, so over dictionaries it fell back to score and
/// then provider ARRIVAL order: <c>select:path,name sort:name-desc limit:3</c> 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 <c>select:</c> list (plus a content sort
/// key) when asked for nodes, so nothing extra is read.</para>
/// </summary>
private static IObservable<QueryResultChange<object>> ProjectAfterMerge(
IObservable<QueryResultChange<MeshNode>> 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<object>
{
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,
};
});

/// <summary>
/// Sort + skip + clip the merged initial set. The authoritative ordering
/// pass for every multi-provider Initial emission. Mirrors the post-collect
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -206,6 +206,22 @@ private IObservable<IReadOnlyList<object>> RunQueryNodes(
bool useSecurityFilter,
bool projectSelect = false,
ReadCompleteness? completeness = null)
=> RunRankedQueryNodes(request, options, useSecurityFilter, projectSelect, completeness)
.Select(ranked => ranked.Rows);

/// <summary>
/// <see cref="RunQueryNodes"/> plus the relevance it ranked by: <c>Scores</c> is one fuzzy
/// score per row, in row order, when a free-text term (and no <c>sort:</c>) made relevance the
/// ordering, and <see langword="null"/> 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).
/// </summary>
private IObservable<(IReadOnlyList<object> Rows, IReadOnlyList<double>? Scores)> RunRankedQueryNodes(
MeshQueryRequest request,
JsonSerializerOptions options,
bool useSecurityFilter,
bool projectSelect,
ReadCompleteness? completeness)
=> CollectMatched(request, options, useSecurityFilter, completeness)
.Select(collected =>
{
Expand Down Expand Up @@ -237,6 +253,9 @@ private IObservable<IReadOnlyList<object>> 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<object> 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<object, double>? relevance = null;
var effectiveOrderBy = parsedQuery.EffectiveOrderBy;
if (effectiveOrderBy != null)
{
Expand All @@ -263,8 +282,16 @@ private IObservable<IReadOnlyList<object>> 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<object, double>(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
Expand All @@ -291,20 +318,24 @@ private IObservable<IReadOnlyList<object>> 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<object> 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<object>)projected.ToList();
IReadOnlyList<object> rows = kept
.Select(node => projectSelect && parsedQuery.Select != null
? ParsedQuery.ProjectToSelect(node, parsedQuery.Select)
: DropUnprojectedContent(node, parsedQuery))
.ToList();
IReadOnlyList<double>? scores = relevance is null
? null
: kept.Select(node => relevance[node]).ToList();
return (rows, scores);
});


Expand Down Expand Up @@ -1609,17 +1640,23 @@ private IObservable<QueryResultChange<T>> ObserveQueryInternal<T>(
// 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<List<(string? Path, T Item)>> RunQuery() =>
RunQueryNodes(request, options, useSecurityFilter,
IObservable<(List<(string? Path, T Item)> Rows, List<double>? 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<double>();
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
Expand Down Expand Up @@ -1706,10 +1743,10 @@ void Route(DataChangeNotification n)

disposables.Add(
RunQuery().Subscribe(
initialResults =>
initial =>
{
var initialItems = new List<T>();
foreach (var (path, item) in initialResults)
foreach (var (path, item) in initial.Rows)
{
initialItems.Add(item);
if (!string.IsNullOrEmpty(path))
Expand Down Expand Up @@ -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
Expand All @@ -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,
Expand Down
Loading
Loading