feat: DAG data-flow timeline — per-operator distribution protocol + playhead UI - #393
Merged
rapids-bot[bot] merged 15 commits intoJul 20, 2026
Conversation
Add a generic protocol for the DAG data-flow-over-time view: per operator
of a query, a binned timeline of a distribution over (FSM state x
downstream-declared dimension) for downstream-declared measures.
- quent-analyzer: DistributionTimelineBuilder, span-weighted aggregation
over opaque (series, measure, state, dimension) keys
- quent-ui: DistributionTimelineRequest/Decl/Series DTOs (ts-rs exported)
- quent-query-engine-ui: DataFlowTimelineResponse (Unsupported | Binned)
- quent-query-engine-analyzer: UiAnalyzer::data_flow_timeline with a
default Unsupported impl so existing analyzers keep compiling
- quent-query-engine-server: POST /api/engines/{id}/timeline/data-flow
- simulator: reference implementation (dimension = memory resource
instance the state uses, measures = tasks/bytes) + functional tests
against the fixed scenario
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Adds the UI half of the data-flow distribution protocol: a time slider
(playhead) under the query-plan DAG that, at time T, shows on every DAG
node a mini stacked bar of entities per FSM state plus a thin bar of the
server-declared dimension breakdown (e.g. data location / memory tier).
Scrubbing animates where data accumulates. Everything is server-declared
via POST /api/engines/{id}/timeline/data-flow — no hardcoded state,
dimension, or measure names in the UI; "Unsupported"/empty responses hide
the feature entirely.
- @quent/utils: re-export the new ts-bindings (DataFlowTimelineResponse,
DistributionDecl/Series, DistributionTimelineRequest, MeasureDecl,
DimensionKeyDecl).
- @quent/client: fetchDataFlow + dataFlowQueryOptions/useDataFlow
(keepPreviousData so zoom refetches don't flicker).
- @quent/hooks: private data-flow atoms (HOOKS-02) + selector hooks,
pure unit-tested helpers (normalize/window/bin-index/windowMax/frame
extraction), and useDataFlowSync which keeps the raw response in
react-query, clamps the playhead into the window, and recomputes the
per-bin frame inside startTransition via store.sub — the host component
never re-renders on scrub.
- @quent/components: DagPlayhead (plain DOM slider: play/pause, pointer
capture + rAF-throttled drag, keyboard slider semantics, synced
crosshair on timeline charts via new broadcastSyncedPointer), NodeFlowBar
(only node-level frame subscriber; window-max-stable width, constant
height when empty), plus DAGControls toggle/measure select, DAGLegend
state+dimension groups, and a state × dimension matrix in
DAGNodeInfoPanel.
- app: QueryPlan wires zoom window → useDataFlow → useDataFlowSync and
renders the playhead between the DAG canvas and the info panel.
Verified against the simulator server (endpoint smoke test through the
vite proxy) with typecheck, lint, build, and 499 vitest tests green.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
… bars Node flow bar improvements for the DAG data-flow overlay: - Render each state segment's value inside its colored segment, width-gated purely from frame data (segment px = value/windowMax * 168px track at ~6px/char + 4px pad) — no DOM measurement, labels hide when the segment is too narrow, and absolute positioning inside overflow-hidden segments means zero layout shift. The state bar grows 6px -> 12px to fit legible 8px labels; text color flips by segment luminance (dark text on light colors, white + subtle shadow on dark) for both themes. - The tiny per-node total now shows EVERY declared measure with data at the playhead bin, joined as e.g. "3.2 · 1.4MiB" (count · bytes), still one right-aligned truncating line that collapses to nbsp when empty. - New compact formatters: formatCompactWithPrefix / formatQuantityCompact (2-3 significant digits, no space, prefix+symbol only: "482", "1.2k", "45MiB") and formatDataFlowValueCompact / fitDataFlowSegmentLabel on top. - extractDataFlowFrame now also returns totalsByMeasure (per-operator totals for all declared measures at the bin, zero measures omitted) in the same cheap per-scrub pass. - vitest.config.ts: mirror vite.config.ts resolve.dedupe — without it the workspace packages load their own jotai copies, so a test <Provider> never scopes @quent/hooks atoms and playhead state leaks between tests through jotai's global default store. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
'121 · 5GB' reads like the decimal '121.5GB'; '121 | 5GB' does not. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Three additions to the DAG data-flow overlay: - Segment-label measure toggle: a new "Bar labels" select in the DAG controls switches the values INSIDE the state-bar segments between the declared measures (e.g. batch count vs bytes) independently of the measure that sizes the bars (null = follow the bar measure). The frame carries labelByState/labelByDimension for the label measure — aliased to the bar-measure arrays when they coincide, so scrub-tick cost is unchanged. Width-gating still checks the rendered text against the bar-measure segment width. - Labeled memory-tier bar: the dimension/tier bar grows from 3px to the same 12px labeled height as the state bar (2px gap, capacity colors vs FSM colors) and renders each tier's total inside its segment with the same width-gating and compact formatting, using the label measure. - Tier selection: chips in the DAG controls (labeled with the server-declared dimension name) choose which dimension keys are represented. The selection filters state-bar widths/labels, the tier bar, the node totals line, totalsByMeasure, windowMax (recomputed over the selection, stable while scrubbing), and the info-panel matrix columns; the legend greys out deselected tiers. The last selected tier cannot be unchecked, and empty/stale selections resolve to "all"; the selection resets when the declared key set changes (query/engine switch). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughThe PR adds an end-to-end data-flow distribution timeline: span-weighted analyzer aggregation, API contracts and endpoint, client fetching and state synchronization, compact formatting, and interactive DAG overlays with controls, legends, matrices, bars, and playhead navigation. ChangesDistribution timeline foundation
Estimated code review effort: 4 (Complex) | ~60 minutes Suggested labels: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 8
Note
Quiet mode is enabled, so only the most important comments were posted inline. Other review comments are grouped below.
🟡 Other comments (4)
examples/simulator/analyzer/src/lib.rs-805-815 (1)
805-815: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winReject every unknown requested measure.
A request such as
["tasks", "bogus"]succeeds and silently ignores"bogus"because one recognized measure makes this condition false. Validate each non-empty requested name before computing the selection.Proposed validation
+ let unknown: Vec<_> = request + .measures + .iter() + .filter(|name| name.as_str() != MEASURE_TASKS && name.as_str() != MEASURE_BYTES) + .collect(); + if !unknown.is_empty() { + return Err(AnalyzerError::InvalidArgument(format!( + "unknown measures {unknown:?}; declared measures are \ + '{MEASURE_TASKS}' and '{MEASURE_BYTES}'" + ))); + } + let want = |name: &str| request.measures.is_empty() || request.measures.iter().any(|m| m == name); let want_tasks = want(MEASURE_TASKS); let want_bytes = want(MEASURE_BYTES); - if !want_tasks && !want_bytes { - return Err(AnalyzerError::InvalidArgument(format!( - "unknown measures {:?}; declared measures are '{MEASURE_TASKS}' and '{MEASURE_BYTES}'", - request.measures - ))); - }Add coverage for a mixed valid/invalid request.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@examples/simulator/analyzer/src/lib.rs` around lines 805 - 815, Update the measure validation near the want closure so every non-empty entry in request.measures must equal MEASURE_TASKS or MEASURE_BYTES; reject any mixed valid/invalid request with the existing InvalidArgument error before computing selections. Preserve the empty-list behavior that requests all measures, and add coverage for a request containing one recognized and one unknown measure.ui/packages/@quent/components/src/dag/DagPlayhead.tsx-88-107 (1)
88-107: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winCancel the pending pointer frame before hiding the crosshair.
A queued
requestAnimationFramecan execute afterhandlePointerEnd, callapplyClientX, and show the crosshair again after release. Cancel and clearrafRefandpendingClientXRefbeforehideSyncedPointer().🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ui/packages/`@quent/components/src/dag/DagPlayhead.tsx around lines 88 - 107, Update handlePointerEnd to cancel any queued requestAnimationFrame, clear rafRef, and clear pendingClientXRef before calling hideSyncedPointer(), preventing the pending handlePointerMove callback from reapplying the crosshair after pointer release.ui/packages/@quent/components/src/dag/DagPlayhead.tsx-118-143 (1)
118-143: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winSynchronize the timeline crosshair during keyboard scrubbing.
Arrow, Home, and End keys update the DAG frame but never call
broadcastSyncedPointer, so keyboard navigation diverges from pointer dragging and playback.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ui/packages/`@quent/components/src/dag/DagPlayhead.tsx around lines 118 - 143, Update handleKeyDown to call broadcastSyncedPointer with the resulting playhead position for Arrow, Home, and End navigation, matching the synchronization performed by pointer dragging and playback. Preserve the existing step directions, boundary times, and preventDefault behavior.ui/packages/@quent/utils/src/formatters.ts-157-195 (1)
157-195: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winRenormalize compact values after rounding crosses a prefix boundary.
Prefix selection happens before rounding, producing outputs such as
1000kor1024KiBinstead of1Mor1MiB.
ui/packages/@quent/utils/src/formatters.ts#L157-L195: advance to the next prefix when the rounded mantissa reaches the current unit’s radix.ui/packages/@quent/utils/src/formatters.test.ts#L484-L504: add SI and IEC boundary tests such as999_999 → 1Mand1_048_575 → 1MiB.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ui/packages/`@quent/utils/src/formatters.ts around lines 157 - 195, Update formatCompactWithPrefix and compactDigits in ui/packages/@quent/utils/src/formatters.ts (lines 157-195) to renormalize after rounding: when the rounded mantissa reaches the current prefix radix, advance to the next SI or IEC prefix and format it as 1M or 1MiB rather than 1000k or 1024KiB. Add corresponding SI and IEC boundary tests in ui/packages/@quent/utils/src/formatters.test.ts (lines 484-504), including 999_999 → 1M and 1_048_575 → 1MiB.
🧹 Nitpick comments (1)
crates/analyzer/src/timeline/binned/distribution.rs (1)
58-89: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDrop the unnecessary
Clonebound fromDistributionTimelineBuilder<'a, S>
The builder never clonesS, andKeyedAggregatoronly needsEq + Hash; removingClonewould keep the API less constrained.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/analyzer/src/timeline/binned/distribution.rs` around lines 58 - 89, The DistributionTimelineBuilder implementation unnecessarily requires S: Clone. Remove Clone from the where clause on DistributionTimelineBuilder and retain only the Eq + Hash bounds used by KeyedAggregator, leaving new, config, try_push, and build behavior unchanged.Source: Path instructions
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@domains/query_engine/ui/src/lib.rs`:
- Line 6: Keep the data_flow module non-public by changing its module
declaration to pub(crate), then explicitly re-export only the intended DTO types
from lib.rs via pub use. Update consumers such as the simulator binding build
script to import those types from the crate root rather than through the
internal module path.
In `@examples/simulator/analyzer/src/lib.rs`:
- Around line 915-932: Update the query response construction around
memory_instance_names and dimension_keys so dimensions are collected only from
tasks selected by the requested query, rather than from the global memory_names
engine model. Track each memory location encountered during query-filtered task
processing, then sort that set and build DimensionKeyDecl entries from it;
retain the DIMENSION_NONE entry.
- Around line 70-71: Replace the sentinel dimension key used by the analyzer’s
no-memory states so it cannot collide with a real resource named “none”; prefer
opaque resource IDs for aggregation and declarations while retaining instance
names in display_name, or explicitly reject that reserved name. Update all
related handling at the DIMENSION_NONE definition and the corresponding logic
around the referenced resource aggregation and declaration sites, preserving
correct no-memory behavior.
In `@ui/packages/`@quent/components/src/dag/DAGChart.tsx:
- Line 363: Update the asynchronous layout effect containing calculateLayout to
discard stale results when flowBarVisible or other dependencies trigger a newer
invocation. Track each invocation with a generation token or cancellation flag,
and check it before committing nodes and edges so only the latest layout result
updates state.
In `@ui/packages/`@quent/components/src/dag/DAGLegend.tsx:
- Around line 82-89: Update the legend item and label className expressions in
DAGLegend to use cn() with their base classes and dimmed-dependent classes,
replacing manual string concatenation while preserving the existing opacity and
line-through behavior.
In `@ui/packages/`@quent/components/src/dag/DAGNodeInfoPanel.tsx:
- Around line 26-138: The PascalCase components violate the
one-component-per-file convention. In
ui/packages/@quent/components/src/dag/DAGNodeInfoPanel.tsx lines 26-138, extract
DataFlowMatrix into its own file and extract or inline ColorDot; update imports
and usages while preserving behavior. In
ui/packages/@quent/components/src/query-plan/NodeFlowBar.tsx lines 35-55,
extract or inline SegmentValueLabel so it is not defined alongside other
components.
In `@ui/packages/`@quent/components/src/dag/DagPlayhead.tsx:
- Line 13: Update the timeline utility import in DagPlayhead.tsx to use the
repository’s @ alias instead of the parent-relative path, while preserving the
existing broadcastSyncedPointer, hideSyncedPointer, and nanosToMs imports.
- Around line 157-184: Update the playback effects in DagPlayhead so playback
stops and the synced pointer is hidden whenever enabled is false or bin metadata
is unavailable. Include enabled and bin in the relevant dependency handling, and
ensure disabling the overlay or losing bin clears the active interval and
prevents further state updates or broadcasts.
---
Other comments:
In `@examples/simulator/analyzer/src/lib.rs`:
- Around line 805-815: Update the measure validation near the want closure so
every non-empty entry in request.measures must equal MEASURE_TASKS or
MEASURE_BYTES; reject any mixed valid/invalid request with the existing
InvalidArgument error before computing selections. Preserve the empty-list
behavior that requests all measures, and add coverage for a request containing
one recognized and one unknown measure.
In `@ui/packages/`@quent/components/src/dag/DagPlayhead.tsx:
- Around line 88-107: Update handlePointerEnd to cancel any queued
requestAnimationFrame, clear rafRef, and clear pendingClientXRef before calling
hideSyncedPointer(), preventing the pending handlePointerMove callback from
reapplying the crosshair after pointer release.
- Around line 118-143: Update handleKeyDown to call broadcastSyncedPointer with
the resulting playhead position for Arrow, Home, and End navigation, matching
the synchronization performed by pointer dragging and playback. Preserve the
existing step directions, boundary times, and preventDefault behavior.
In `@ui/packages/`@quent/utils/src/formatters.ts:
- Around line 157-195: Update formatCompactWithPrefix and compactDigits in
ui/packages/@quent/utils/src/formatters.ts (lines 157-195) to renormalize after
rounding: when the rounded mantissa reaches the current prefix radix, advance to
the next SI or IEC prefix and format it as 1M or 1MiB rather than 1000k or
1024KiB. Add corresponding SI and IEC boundary tests in
ui/packages/@quent/utils/src/formatters.test.ts (lines 484-504), including
999_999 → 1M and 1_048_575 → 1MiB.
---
Nitpick comments:
In `@crates/analyzer/src/timeline/binned/distribution.rs`:
- Around line 58-89: The DistributionTimelineBuilder implementation
unnecessarily requires S: Clone. Remove Clone from the where clause on
DistributionTimelineBuilder and retain only the Eq + Hash bounds used by
KeyedAggregator, leaving new, config, try_push, and build behavior unchanged.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: QUIET
Plan: Enterprise
Run ID: 1a2a1ba8-a5fb-4518-93d2-4030305b23b7
⛔ Files ignored due to path filters (7)
examples/simulator/server/ts-bindings/DataFlowTimelineBinned.tsis excluded by!examples/simulator/server/ts-bindings/**examples/simulator/server/ts-bindings/DataFlowTimelineResponse.tsis excluded by!examples/simulator/server/ts-bindings/**examples/simulator/server/ts-bindings/DimensionKeyDecl.tsis excluded by!examples/simulator/server/ts-bindings/**examples/simulator/server/ts-bindings/DistributionDecl.tsis excluded by!examples/simulator/server/ts-bindings/**examples/simulator/server/ts-bindings/DistributionSeries.tsis excluded by!examples/simulator/server/ts-bindings/**examples/simulator/server/ts-bindings/DistributionTimelineRequest.tsis excluded by!examples/simulator/server/ts-bindings/**examples/simulator/server/ts-bindings/MeasureDecl.tsis excluded by!examples/simulator/server/ts-bindings/**
📒 Files selected for processing (37)
crates/analyzer/src/timeline/binned/distribution.rscrates/analyzer/src/timeline/binned/mod.rscrates/ui/src/timeline/distribution.rscrates/ui/src/timeline/mod.rsdocs/domains/query_engine/README.mddomains/query_engine/analyzer/src/ui.rsdomains/query_engine/server/src/ui.rsdomains/query_engine/tests/fixed/tests/data_flow.rsdomains/query_engine/ui/src/data_flow.rsdomains/query_engine/ui/src/lib.rsexamples/simulator/analyzer/src/lib.rsexamples/simulator/server/build.rsui/packages/@quent/client/src/api.tsui/packages/@quent/client/src/dataFlow.tsui/packages/@quent/client/src/index.tsui/packages/@quent/components/src/dag/DAGChart.tsxui/packages/@quent/components/src/dag/DAGControls.tsxui/packages/@quent/components/src/dag/DAGLegend.tsxui/packages/@quent/components/src/dag/DAGNodeInfoPanel.tsxui/packages/@quent/components/src/dag/DagPlayhead.tsxui/packages/@quent/components/src/index.tsui/packages/@quent/components/src/lib/timeline.utils.tsui/packages/@quent/components/src/query-plan/NodeFlowBar.tsxui/packages/@quent/components/src/query-plan/QueryPlanNode.tsxui/packages/@quent/hooks/src/atoms/dataFlow.tsui/packages/@quent/hooks/src/dataFlow/dataFlow.utils.test.tsui/packages/@quent/hooks/src/dataFlow/dataFlow.utils.tsui/packages/@quent/hooks/src/dataFlow/dataFlowSelectors.tsui/packages/@quent/hooks/src/dataFlow/useDataFlowSync.tsui/packages/@quent/hooks/src/index.tsui/packages/@quent/utils/src/formatters.test.tsui/packages/@quent/utils/src/formatters.tsui/packages/@quent/utils/src/index.tsui/packages/@quent/utils/src/types/index.tsui/src/components/DataFlowOverlay.test.tsxui/src/components/QueryPlan.tsxui/vitest.config.ts
Comment on lines
+26
to
+138
| const ColorDot = ({ color }: { color: string }) => ( | ||
| <span className="inline-block h-2 w-2 rounded-sm shrink-0" style={{ backgroundColor: color }} /> | ||
| ); | ||
|
|
||
| /** | ||
| * State × dimension matrix of the data-flow distribution for the selected | ||
| * operator at the playhead's bin. Values are span-weighted per-bin averages | ||
| * ("during this bin"), so fractional counts are expected. Columns are | ||
| * filtered to the SELECTED dimension keys (tiers) — deselected tiers are | ||
| * zero in the frame anyway, so hiding their columns loses nothing. | ||
| */ | ||
| const DataFlowMatrix = ({ | ||
| meta, | ||
| frame, | ||
| operatorFrame, | ||
| isDark, | ||
| }: { | ||
| meta: DataFlowMeta; | ||
| frame: DataFlowFrame; | ||
| operatorFrame: DataFlowOperatorFrame; | ||
| isDark: boolean; | ||
| }) => { | ||
| const paletteTheme: PaletteTheme = isDark ? 'dark' : 'light'; | ||
| const allDimensionKeys = meta.decl.dimension_keys; | ||
| // Keep original decl-order indices — the frame's matrix/byDimension are | ||
| // indexed by declaration order, not by the filtered column order. | ||
| const dimensionColumns = useMemo( | ||
| () => | ||
| allDimensionKeys | ||
| .map((key, index) => ({ key, index })) | ||
| .filter(({ key }) => meta.dimensionSelection.has(key.key)), | ||
| [allDimensionKeys, meta.dimensionSelection] | ||
| ); | ||
| const stateColor = useMemo( | ||
| () => | ||
| createFsmTypeColorFn(meta.fsmType ? { [meta.fsmType.name]: meta.fsmType } : {}, paletteTheme), | ||
| [meta, paletteTheme] | ||
| ); | ||
| const dimensionColor = useMemo( | ||
| () => | ||
| createCapacitiesColorFn( | ||
| allDimensionKeys.map(k => k.key), | ||
| paletteTheme | ||
| ), | ||
| [allDimensionKeys, paletteTheme] | ||
| ); | ||
|
|
||
| const fmt = (value: number) => formatDataFlowValue(value, frame.measure, meta); | ||
| const measureDecl = meta.decl.measures.find(m => m.name === frame.measure); | ||
|
|
||
| return ( | ||
| <div className="pt-1.5"> | ||
| <div className="text-xs font-medium"> | ||
| Data flow @ <DataText>{formatDuration(frame.timeS * 1000)}</DataText> | ||
| <span className="text-muted-foreground font-normal"> | ||
| {' '} | ||
| · {measureDecl?.display_name ?? frame.measure} during this bin | ||
| </span> | ||
| </div> | ||
| <table className="mt-1 text-xs w-full border-separate border-spacing-0"> | ||
| <thead> | ||
| <tr> | ||
| <th className="text-left font-normal text-muted-foreground pr-2"> | ||
| {meta.decl.dimension_name} | ||
| </th> | ||
| {dimensionColumns.map(({ key: k }) => ( | ||
| <th key={k.key} className="text-right font-normal text-muted-foreground px-1.5"> | ||
| <span className="inline-flex items-center gap-1"> | ||
| <ColorDot color={dimensionColor(k.key)} /> | ||
| <DataText>{k.display_name}</DataText> | ||
| </span> | ||
| </th> | ||
| ))} | ||
| <th className="text-right font-medium text-muted-foreground pl-1.5">Total</th> | ||
| </tr> | ||
| </thead> | ||
| <tbody> | ||
| {meta.stateNames.map((state, stateIndex) => ( | ||
| <tr key={state}> | ||
| <td className="pr-2"> | ||
| <span className="inline-flex items-center gap-1"> | ||
| <ColorDot color={stateColor(state)} /> | ||
| <DataText>{state}</DataText> | ||
| </span> | ||
| </td> | ||
| {dimensionColumns.map(({ key: k, index: dimensionIndex }) => ( | ||
| <td key={k.key} className="text-right px-1.5 text-muted-foreground"> | ||
| <DataText> | ||
| {fmt(operatorFrame.matrix[stateIndex]?.[dimensionIndex] ?? 0)} | ||
| </DataText> | ||
| </td> | ||
| ))} | ||
| <td className="text-right pl-1.5"> | ||
| <DataText>{fmt(operatorFrame.byState[stateIndex] ?? 0)}</DataText> | ||
| </td> | ||
| </tr> | ||
| ))} | ||
| <tr> | ||
| <td className="pr-2 pt-0.5 font-medium">Total</td> | ||
| {dimensionColumns.map(({ key: k, index: dimensionIndex }) => ( | ||
| <td key={k.key} className="text-right px-1.5 pt-0.5"> | ||
| <DataText>{fmt(operatorFrame.byDimension[dimensionIndex] ?? 0)}</DataText> | ||
| </td> | ||
| ))} | ||
| <td className="text-right pl-1.5 pt-0.5 font-medium"> | ||
| <DataText>{fmt(operatorFrame.total)}</DataText> | ||
| </td> | ||
| </tr> | ||
| </tbody> | ||
| </table> | ||
| </div> | ||
| ); | ||
| }; |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Keep each PascalCase component in its own file.
ui/packages/@quent/components/src/dag/DAGNodeInfoPanel.tsx#L26-L138: extractDataFlowMatrix; extract or inlineColorDot.ui/packages/@quent/components/src/query-plan/NodeFlowBar.tsx#L35-L55: extract or inlineSegmentValueLabel.
As per path instructions, “Components PascalCase one-per-file.”
📍 Affects 2 files
ui/packages/@quent/components/src/dag/DAGNodeInfoPanel.tsx#L26-L138(this comment)ui/packages/@quent/components/src/query-plan/NodeFlowBar.tsx#L35-L55
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@ui/packages/`@quent/components/src/dag/DAGNodeInfoPanel.tsx around lines 26 -
138, The PascalCase components violate the one-component-per-file convention. In
ui/packages/@quent/components/src/dag/DAGNodeInfoPanel.tsx lines 26-138, extract
DataFlowMatrix into its own file and extract or inline ColorDot; update imports
and usages while preserving behavior. In
ui/packages/@quent/components/src/query-plan/NodeFlowBar.tsx lines 35-55,
extract or inline SegmentValueLabel so it is not defined alongside other
components.
Source: Path instructions
| usePlayheadTimeS, | ||
| useSetPlayheadTimeS, | ||
| } from '@quent/hooks'; | ||
| import { broadcastSyncedPointer, hideSyncedPointer, nanosToMs } from '../lib/timeline.utils'; |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Use the configured alias instead of a parent-relative import.
Import the timeline utilities through the repository’s @ alias.
As per path instructions, “use the @ alias over deep relative imports.”
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@ui/packages/`@quent/components/src/dag/DagPlayhead.tsx at line 13, Update the
timeline utility import in DagPlayhead.tsx to use the repository’s @ alias
instead of the parent-relative path, while preserving the existing
broadcastSyncedPointer, hideSyncedPointer, and nanosToMs imports.
Source: Path instructions
- Curate the domain UI crate's public surface: data_flow module is private, DataFlowTimelineBinned/DataFlowTimelineResponse re-exported from the crate root (consumers updated). - Simulator data_flow_timeline: declare only dimension keys observed in the requested query's tasks, and grow the no-memory sentinel key until it cannot collide with a real memory resource name. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
- DagPlayhead: stop playback and hide the synced crosshair when the overlay is disabled or bin metadata disappears (the component stays mounted while rendering null, so the play interval kept ticking); covered by a new fake-timer test. - DAGChart: discard stale async layout results with a cancelled flag so interleaved calculateLayout calls cannot overwrite a newer layout. - DAGLegend: merge conditional classes via cn() instead of template string concatenation. - DAGNodeInfoPanel: extract DataFlowMatrix and ColorDot into their own files (one PascalCase component per file). - NodeFlowBar: extract SegmentValueLabel into its own file. DagPlayhead's relative '../lib/timeline.utils' import is kept as-is: the '@' alias maps to the app's src/ only, and relative imports are the package-wide convention in @quent/components. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Contributor
Author
|
Addressed the CodeRabbit review in the two commits just pushed: Rust (
UI (
Skipped (1): the All suites green: 🤖 Generated with Claude Code |
There was a problem hiding this comment.
Actionable comments posted: 1
Note
Quiet mode is enabled, so only the most important comments were posted inline. Other review comments are grouped below.
🟡 Other comments (1)
ui/packages/@quent/components/src/dag/DagPlayhead.tsx-157-165 (1)
157-165: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winCancel queued scrubbing frames when hiding the overlay.
A
requestAnimationFramequeued by pointer movement survives this effect and can callapplyClientXafter the component rendersnull, re-broadcasting the crosshair. CancelrafRef.currentand clearpendingClientXRef.currenthere; add a regression test that queues a move before disabling.As per path instructions, “Check hook dependency arrays and effect cleanup.”
Proposed fix
useEffect(() => { if (enabled && bin) return; + if (rafRef.current != null) { + cancelAnimationFrame(rafRef.current); + rafRef.current = null; + } + pendingClientXRef.current = null; setIsPlaying(false); hideSyncedPointer(); }, [enabled, bin]);🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ui/packages/`@quent/components/src/dag/DagPlayhead.tsx around lines 157 - 165, Update the playback-stopping useEffect in DagPlayhead to cancel any queued requestAnimationFrame via rafRef.current and clear pendingClientXRef.current when the overlay is disabled or bin metadata is unavailable. Preserve the existing setIsPlaying and hideSyncedPointer behavior, include the relevant hook dependencies, and add a regression test covering a queued pointer move followed by disabling the overlay.Source: Path instructions
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@ui/packages/`@quent/components/src/dag/DataFlowMatrix.tsx:
- Around line 75-123: Update the matrix table in DataFlowMatrix so the
state-label cells use semantic row headers and the top-left header identifies
the state axis rather than the dimension axis. Preserve the existing column
headers and displayed labels while using appropriate header scope attributes to
associate every value with both the state and dimension axes.
---
Other comments:
In `@ui/packages/`@quent/components/src/dag/DagPlayhead.tsx:
- Around line 157-165: Update the playback-stopping useEffect in DagPlayhead to
cancel any queued requestAnimationFrame via rafRef.current and clear
pendingClientXRef.current when the overlay is disabled or bin metadata is
unavailable. Preserve the existing setIsPlaying and hideSyncedPointer behavior,
include the relevant hook dependencies, and add a regression test covering a
queued pointer move followed by disabling the overlay.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: QUIET
Plan: Enterprise
Run ID: 1e85fe62-6201-4e89-be2a-482eaefc5f44
📒 Files selected for processing (15)
domains/query_engine/analyzer/src/ui.rsdomains/query_engine/server/src/ui.rsdomains/query_engine/tests/fixed/tests/data_flow.rsdomains/query_engine/ui/src/lib.rsexamples/simulator/analyzer/src/lib.rsexamples/simulator/server/build.rsui/packages/@quent/components/src/dag/ColorDot.tsxui/packages/@quent/components/src/dag/DAGChart.tsxui/packages/@quent/components/src/dag/DAGLegend.tsxui/packages/@quent/components/src/dag/DAGNodeInfoPanel.tsxui/packages/@quent/components/src/dag/DagPlayhead.tsxui/packages/@quent/components/src/dag/DataFlowMatrix.tsxui/packages/@quent/components/src/query-plan/NodeFlowBar.tsxui/packages/@quent/components/src/query-plan/SegmentValueLabel.tsxui/src/components/DataFlowOverlay.test.tsx
felipeblazing
added a commit
to felipeblazing/sirius
that referenced
this pull request
Jul 15, 2026
- Point the temporary [patch] at the quent fork branch (felipeblazing/quent claude/dag-data-flow-timeline-d6f6e4, PR rapidsai/quent#393) instead of a local worktree path so CI can build. - Apply clang-format/include-order to the batch telemetry files (the lint job runs pre-commit on all files). - Import DataFlowTimeline types from the quent-query-engine-ui crate root, matching the curated re-exports quent adopted in review. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
- cargo fmt over the new distribution/simulator/test files (CI's fmt --check gate). - DataFlowMatrix: semantic scope=col/row headers, state cells as row headers, and a 'State / <dimension>' top-left header so screen readers associate each value with both axes (review feedback). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
johanpel
reviewed
Jul 16, 2026
johanpel
left a comment
johanpel
left a comment
Contributor
There was a problem hiding this comment.
Inspected the Rust side. Thanks @felipeblazing, this is a good start and an exciting feature for the QE parts.
Comment on lines
+808
to
+810
| let want_tasks = want(MEASURE_TASKS); | ||
| let want_bytes = want(MEASURE_BYTES); | ||
| if !want_tasks && !want_bytes { |
Contributor
There was a problem hiding this comment.
If either of these is valid we accept unknown measures, should we error instead?
Comment on lines
+192
to
+216
| ## Data-flow distribution timeline | ||
|
|
||
| The analyzer trait offers an optional `data_flow_timeline` method (HTTP: | ||
| `POST /api/engines/{engine_id}/timeline/data-flow`) powering the UI's | ||
| data-flow-over-time view of a query plan: for every [Operator][operator] of a | ||
| query, a binned timeline of a distribution over | ||
| (FSM state × application-defined dimension), for one or more | ||
| application-declared measures. | ||
|
|
||
| Consistent with Operators having no FSM (see [Operator][operator] notes), all | ||
| semantics live in the application's analyzer: | ||
|
|
||
| - **Entity**: which FSM type is distributed (e.g. a task or batch entity that | ||
| works on behalf of an Operator), referenced by `entity_type_name` into the | ||
| query bundle's FSM type declarations for state names and ordering. | ||
| - **Dimension**: an opaque, small, enumerable key set declared per response | ||
| (e.g. where an entity's data resides), with display names and stable order. | ||
| - **Measures**: named weights (e.g. an entity count, resident bytes) with a | ||
| quantity spec reference for unit formatting. | ||
|
|
||
| Bin values are span-weighted (an entity in a state for a fraction of a bin | ||
| contributes that fraction), matching all other timelines. Analyzers that do | ||
| not provide the feature return `Unsupported` — the default implementation — | ||
| and the UI hides the view. | ||
|
|
Contributor
There was a problem hiding this comment.
The intention of this doc is to specify and explain the query engine domain event model, but not to explain the UI or any implementation details (those are not in the current docs/ at all). Let's find a better place for this or remove and create an issue so we can address it when we have some introductory sections / scaffolding in place.
Comment on lines
+28
to
+39
| /// Identity of one aggregation cell of a distribution timeline. | ||
| #[derive(Clone, Debug, PartialEq, Eq, Hash)] | ||
| pub struct DistributionKey<'a, S> { | ||
| /// Opaque series the sample belongs to (e.g. an operator id downstream). | ||
| pub series: S, | ||
| /// The measure this weight contributes to (e.g. an entity count). | ||
| pub measure: &'a str, | ||
| /// The FSM state name during the span. | ||
| pub state: &'a str, | ||
| /// Application-defined dimension key (opaque to the aggregation). | ||
| pub dimension: &'a str, | ||
| } |
Contributor
There was a problem hiding this comment.
I think it would be nice to make this more generic:
Suggested change
| /// Identity of one aggregation cell of a distribution timeline. | |
| #[derive(Clone, Debug, PartialEq, Eq, Hash)] | |
| pub struct DistributionKey<'a, S> { | |
| /// Opaque series the sample belongs to (e.g. an operator id downstream). | |
| pub series: S, | |
| /// The measure this weight contributes to (e.g. an entity count). | |
| pub measure: &'a str, | |
| /// The FSM state name during the span. | |
| pub state: &'a str, | |
| /// Application-defined dimension key (opaque to the aggregation). | |
| pub dimension: &'a str, | |
| } | |
| /// Identity of one aggregation cell of a distribution timeline. | |
| #[derive(Clone, Debug, PartialEq, Eq, Hash)] | |
| pub struct DistributionKey<S, M, St, D> { | |
| /// Opaque series the sample belongs to (e.g. an operator id downstream). | |
| pub series: S, | |
| /// The measure this weight contributes to (e.g. an entity count). | |
| pub measure: M, | |
| /// The FSM state name during the span. | |
| pub state: St, | |
| /// Application-defined dimension key (opaque to the aggregation). | |
| pub dimension: D, | |
| } |
?
This way the DistributionTimelineBuilder would only require Eq + Hash bounds and we wouldn't always need stringification.
Comment on lines
+91
to
+194
| use super::*; | ||
|
|
||
| fn test_config() -> BinnedSpan { | ||
| BinnedSpan::try_new( | ||
| SpanNanoSec::try_new(0, 1000).unwrap(), | ||
| NonZero::try_from(10).unwrap(), | ||
| ) | ||
| .unwrap() | ||
| } | ||
|
|
||
| fn key<'a>( | ||
| series: u32, | ||
| measure: &'a str, | ||
| state: &'a str, | ||
| dimension: &'a str, | ||
| ) -> DistributionKey<'a, u32> { | ||
| DistributionKey { | ||
| series, | ||
| measure, | ||
| state, | ||
| dimension, | ||
| } | ||
| } | ||
|
|
||
| #[test] | ||
| fn span_weighting_across_bin_boundaries() -> AnalyzerResult<()> { | ||
| let mut builder = DistributionTimelineBuilder::new(test_config()); | ||
|
|
||
| // Spans [0, 300) and [250, 450) of weight 1 each. | ||
| builder.try_push(key(1, "count", "a", "x"), SpanNanoSec::try_new(0, 300).unwrap(), 1.0)?; | ||
| builder.try_push( | ||
| key(1, "count", "a", "x"), | ||
| SpanNanoSec::try_new(250, 450).unwrap(), | ||
| 1.0, | ||
| )?; | ||
|
|
||
| let timeline = builder.build(); | ||
| let bins = timeline.data.get(&key(1, "count", "a", "x")).unwrap(); | ||
| assert_eq!( | ||
| bins[..], | ||
| [1.0, 1.0, 1.5, 1.0, 0.5, 0.0, 0.0, 0.0, 0.0, 0.0] | ||
| ); | ||
| Ok(()) | ||
| } | ||
|
|
||
| #[test] | ||
| fn distinct_series_measures_states_dimensions() -> AnalyzerResult<()> { | ||
| let mut builder = DistributionTimelineBuilder::new(test_config()); | ||
| let span = SpanNanoSec::try_new(0, 1000).unwrap(); | ||
|
|
||
| builder.try_push(key(1, "count", "a", "x"), span, 1.0)?; | ||
| builder.try_push(key(2, "count", "a", "x"), span, 1.0)?; | ||
| builder.try_push(key(1, "bytes", "a", "x"), span, 100.0)?; | ||
| builder.try_push(key(1, "count", "b", "x"), span, 1.0)?; | ||
| builder.try_push(key(1, "count", "a", "y"), span, 1.0)?; | ||
|
|
||
| let timeline = builder.build(); | ||
| assert_eq!(timeline.data.len(), 5); | ||
| assert_eq!( | ||
| timeline.data.get(&key(1, "bytes", "a", "x")).unwrap()[..], | ||
| [100.0; 10] | ||
| ); | ||
| assert_eq!( | ||
| timeline.data.get(&key(2, "count", "a", "x")).unwrap()[..], | ||
| [1.0; 10] | ||
| ); | ||
| Ok(()) | ||
| } | ||
|
|
||
| #[test] | ||
| fn zero_duration_span_is_noop() -> AnalyzerResult<()> { | ||
| let mut builder = DistributionTimelineBuilder::new(test_config()); | ||
| builder.try_push( | ||
| key(1, "count", "a", "x"), | ||
| SpanNanoSec::try_new(500, 500).unwrap(), | ||
| 1.0, | ||
| )?; | ||
|
|
||
| let timeline = builder.build(); | ||
| // The key exists (aggregator was created) but all bins remain zero. | ||
| let bins = timeline.data.get(&key(1, "count", "a", "x")).unwrap(); | ||
| assert_eq!(bins[..], [0.0; 10]); | ||
| Ok(()) | ||
| } | ||
|
|
||
| #[test] | ||
| fn out_of_window_span_contributes_nothing() -> AnalyzerResult<()> { | ||
| let mut builder = DistributionTimelineBuilder::new(test_config()); | ||
| builder.try_push( | ||
| key(1, "count", "a", "x"), | ||
| SpanNanoSec::try_new(2000, 3000).unwrap(), | ||
| 1.0, | ||
| )?; | ||
|
|
||
| let timeline = builder.build(); | ||
| let bins = timeline.data.get(&key(1, "count", "a", "x")).unwrap(); | ||
| assert_eq!(bins[..], [0.0; 10]); | ||
| Ok(()) | ||
| } | ||
| } |
Contributor
There was a problem hiding this comment.
The tests here seem a bit superfluous as they test functionality of the inner aggregator more than any functionality expressed in this source. If these tests do not exist on the aggregator yet we should move them there.
Comment on lines
+4
to
+15
| //! Binned timelines of weighted distributions over (state, dimension) pairs. | ||
| //! | ||
| //! A distribution timeline describes, per opaque series (e.g. an operator in a | ||
| //! query engine), how some weighted quantity (a "measure", e.g. an entity | ||
| //! count) is distributed over the states of a finite state machine and an | ||
| //! application-defined dimension (e.g. which resource holds the entity's | ||
| //! data), for each time bin of a window. | ||
| //! | ||
| //! This module is application-agnostic: series, measures, states, and | ||
| //! dimension keys are all opaque to the aggregation. Downstream analyzers | ||
| //! decide what they mean and are expected to keep dimension keys a small | ||
| //! enumerable set. |
Contributor
There was a problem hiding this comment.
Nit: I was thrown off a bit about what this does because "distribution" might imply a probability distribution, but the values within each bin are absolute bin time weighted measures and not normalized to sum to 1. Perhaps we could name this "categorical" or something?
Comment on lines
+32
to
+34
| /// This analyzer does not provide data-flow distributions; the UI hides | ||
| /// the corresponding view. | ||
| Unsupported, |
Contributor
There was a problem hiding this comment.
Perhaps we can add an Unsupported variant to the AnalyzerError instead? That could be leveraged by other endpoints too.
States the analyzer appends beyond the FSM declaration (e.g. a working- space series) fell back to hash-based colors that could collide with a declared state's palette color (batch_queued vs task_working_space). createDataFlowStateColorFn keeps declared states on their declaration- index colors (consistent with timeline lanes) and assigns appended states the palette slots after the declared block. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
DistributionDecl gains default_measure so the downstream analyzer can pick which measure the UI selects initially (e.g. bytes rather than count); None keeps the first declared measure. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
There was a problem hiding this comment.
Note
Quiet mode is enabled, so only the most important comments were posted inline. Other review comments are grouped below.
🟡 Other comments (1)
ui/packages/@quent/utils/src/colors.ts-219-229 (1)
219-229: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winSynthetic states can collide with declared colors once the palette fills up.
getColorByIndexwraps modulo the palette length, so whendeclared.size >= palette.lengththe first appended state reuses an existing declared-state color. Add an explicit overflow policy or guard, and cover the boundary case with a test.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ui/packages/`@quent/utils/src/colors.ts around lines 219 - 229, The state-color indexing in the returned resolver can assign appended states an index that wraps onto declared-state colors when the palette is exhausted. Update the surrounding color-resolution logic to apply an explicit overflow policy or guard when assigning appended indices, preserving declared colors and the existing fallback behavior; add a test covering declared states at or beyond the palette length and the first appended state.
🧹 Nitpick comments (1)
ui/packages/@quent/utils/src/colors.test.ts (1)
473-483: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueTest fixture bypasses type-checking via
as never.Casting the fixture
as nevermeans the compiler never verifies it actually satisfiesFsmTypeDecl; a future rename/shape change in the real type (or in theusagesfield) won't be caught here. Consider using a properly-typed literal (or a minimalPartial<FsmTypeDecl>-safe builder) instead.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ui/packages/`@quent/utils/src/colors.test.ts around lines 473 - 483, Replace the fsmType fixture’s as never cast with a properly typed literal or the existing minimal FsmTypeDecl-safe builder, ensuring name, states, transitions, and each state’s usages satisfy the real type definition. Keep the fixture’s intended batch state data unchanged while restoring compile-time validation.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Other comments:
In `@ui/packages/`@quent/utils/src/colors.ts:
- Around line 219-229: The state-color indexing in the returned resolver can
assign appended states an index that wraps onto declared-state colors when the
palette is exhausted. Update the surrounding color-resolution logic to apply an
explicit overflow policy or guard when assigning appended indices, preserving
declared colors and the existing fallback behavior; add a test covering declared
states at or beyond the palette length and the first appended state.
---
Nitpick comments:
In `@ui/packages/`@quent/utils/src/colors.test.ts:
- Around line 473-483: Replace the fsmType fixture’s as never cast with a
properly typed literal or the existing minimal FsmTypeDecl-safe builder,
ensuring name, states, transitions, and each state’s usages satisfy the real
type definition. Keep the fixture’s intended batch state data unchanged while
restoring compile-time validation.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: QUIET
Plan: Enterprise
Run ID: 3faad64e-9ade-45d9-a484-1c1cb030aa6b
📒 Files selected for processing (9)
crates/analyzer/src/timeline/binned/distribution.rsdomains/query_engine/tests/fixed/tests/data_flow.rsexamples/simulator/analyzer/src/lib.rsui/packages/@quent/components/src/dag/DAGLegend.tsxui/packages/@quent/components/src/dag/DataFlowMatrix.tsxui/packages/@quent/components/src/query-plan/NodeFlowBar.tsxui/packages/@quent/utils/src/colors.test.tsui/packages/@quent/utils/src/colors.tsui/packages/@quent/utils/src/index.ts
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
examples/simulator/analyzer/src/lib.rs (1)
805-815: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winReject unknown measures even when another requested measure is valid.
["tasks", "unknown"]currently succeeds because validation only errors when none of the requested names are recognized. This silently returns partial data and hides client/protocol mistakes. Validate every requested name before computingwant_tasksandwant_bytes.Suggested fix
let config = request.config.try_into_binned_span(epoch)?; +if request + .measures + .iter() + .any(|measure| measure != MEASURE_TASKS && measure != MEASURE_BYTES) +{ + return Err(AnalyzerError::InvalidArgument(format!( + "unknown measures {:?}; declared measures are '{MEASURE_TASKS}' and '{MEASURE_BYTES}'", + request.measures + ))); +} + // Which of the declared measures to compute; empty means all.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@examples/simulator/analyzer/src/lib.rs` around lines 805 - 815, Update measure validation near the want, want_tasks, and want_bytes closures to reject any request.measures entry not equal to MEASURE_TASKS or MEASURE_BYTES, including mixed valid and unknown lists. Perform this validation before computing the requested measure flags, while preserving the empty-list behavior that requests all measures.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@examples/simulator/analyzer/src/lib.rs`:
- Around line 805-815: Update measure validation near the want, want_tasks, and
want_bytes closures to reject any request.measures entry not equal to
MEASURE_TASKS or MEASURE_BYTES, including mixed valid and unknown lists. Perform
this validation before computing the requested measure flags, while preserving
the empty-list behavior that requests all measures.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: QUIET
Plan: Enterprise
Run ID: 3223da45-13e1-47af-a8bf-e6ade29124ae
⛔ Files ignored due to path filters (1)
examples/simulator/server/ts-bindings/DistributionDecl.tsis excluded by!examples/simulator/server/ts-bindings/**
📒 Files selected for processing (2)
crates/ui/src/timeline/distribution.rsexamples/simulator/analyzer/src/lib.rs
resolveDataFlowMeasure now prefers the analyzer-declared
decl.default_measure when the user has no (valid) selection, so the DAG
flow bars start on the measure the analyzer picked (Sirius declares
"bytes" — bars open on Batch bytes). An explicit valid selection still
wins, and an absent/undeclared default keeps the first-declared-measure
fallback; the label measure follows the resolved bar measure unchanged.
extractDataFlowFrame additionally accumulates dimensionTotalsByMeasure
in its existing single pass: global per-tier totals at the bin for every
declared measure, summed over ALL operators, states, and dimension keys
(deliberately unfiltered by the tier selection). DAGLegend renders each
tier of the dimension group with its total at the playhead bin in the
current flow measure ("GPU-0 · 12.4GiB"), formatted via the measure's
quantity spec; zero totals get no suffix and dimmed (deselected) tiers
keep theirs. The tier group lives in a memoized leaf that alone
subscribes to the per-scrub frame, so the rest of the legend does not
re-render per tick.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@ui/packages/`@quent/components/src/dag/DAGLegend.tsx:
- Line 4: Move the DataFlowTierLegend component from DAGLegend.tsx into a new
DataFlowTierLegend.tsx file, including only the frame-specific imports and
dependencies it uses. Remove its implementation and no-longer-needed imports
from DAGLegend.tsx, while preserving its export and existing behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: QUIET
Plan: Enterprise
Run ID: 2ff457e2-d59d-4974-a299-687c37009127
📒 Files selected for processing (4)
ui/packages/@quent/components/src/dag/DAGLegend.tsxui/packages/@quent/hooks/src/dataFlow/dataFlow.utils.test.tsui/packages/@quent/hooks/src/dataFlow/dataFlow.utils.tsui/src/components/DataFlowOverlay.test.tsx
…tocol - Rename the 'distribution' aggregation to 'categorical' (values are absolute time-weighted quantities, not a probability distribution): CategoricalKey/Timeline/Builder, CategoricalTimelineRequest/Decl/ Series, regenerated ts-bindings. - CategoricalKey is generic over <S, M, St, D> (Eq + Hash only) — no forced stringification. - Replace the response-level Unsupported variant with AnalyzerError::Unsupported, mapped to HTTP 501 by the server; the data-flow endpoint now returns DataFlowTimelineBinned directly and the trait default errs Unsupported. - Reject any unknown requested measure, even alongside valid ones (simulator reference impl + test). - Move span-weighting/zero-span/out-of-window tests down to the keyed aggregator where that behavior lives; the categorical layer keeps key-identity and non-string-key tests. - Drop the UI-protocol section from the domain event-model doc (tracked separately for future analyzer-protocol docs). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Follow-up to the Rust-side maintainer review on PR rapidsai#393, UI only: - Rename distribution -> categorical: re-export CategoricalDecl, CategoricalSeries and CategoricalTimelineRequest from @quent/utils (the Distribution* bindings are gone) and update all usages in the client fetcher, data-flow utils/hooks and tests. MeasureDecl and DimensionKeyDecl are unchanged. - Drop the DataFlowTimelineResponse enum: the endpoint now returns DataFlowTimelineBinned directly and signals unsupported analyzers with HTTP 501. fetchDataFlow resolves the 501 to a null sentinel instead of throwing, so react-query settles as "unavailable" without retries; normalizeDataFlowResponse/isDataFlowAvailable lose the "Unsupported"/{Binned: ...} cases and tests now exercise the bare binned object plus the null path (new api.test.ts covers 501 -> null, other errors still throw). - Extract DataFlowTierLegend out of DAGLegend.tsx into its own DataFlowTierLegend.tsx with its frame-specific imports, behavior unchanged (CategoricalLegend is now exported for reuse). typecheck, test:run (583), lint and build are all green. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Contributor
Author
|
Thanks for the review @johanpel — all six points addressed in the latest pushes:
Full suites green: 🤖 Generated with Claude Code |
johanpel
approved these changes
Jul 20, 2026
johanpel
left a comment
johanpel
left a comment
Contributor
There was a problem hiding this comment.
Thanks for addressing all the comments, approving Rust changes.
Contributor
Author
|
/merge |
felipeblazing
added a commit
to felipeblazing/sirius
that referenced
this pull request
Jul 20, 2026
is merged Bump the quent git revs from the pre-protocol pin to the merge commit of rapidsai/quent#393 and drop the temporary [patch] section that pointed the 16 quent crates at the development fork branch. Adapt to the quent-attributes -> quent-dynamic-attributes rename (rapidsai/quent#418) that landed on quent main in the meantime, and add the origin_tier/input_bytes fields (from the dev merge) to the analyzer test's Preparing fixture. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
rapids-bot Bot
pushed a commit
that referenced
this pull request
Jul 21, 2026
#393 merged generated ts bindings that weren't formatted using the changes of #421, this fixes them. cc. @johallar @cmatzenbach could we please consider #255 to avoid these issues altogether? Authors: - Johan Peltenburg (https://github.com/johanpel) Approvers: - Matthijs Brobbel (https://github.com/mbrobbel) - Dhruv Vats (https://github.com/dhruv9vats) URL: #424
kevkrist
pushed a commit
to kevkrist/sirius
that referenced
this pull request
Jul 24, 2026
…ew (sirius-db#1187) ## What Instruments **batch placements** — the lifecycle of each data batch published to a consuming pipeline's input port — and implements quent's new per-operator data-flow distribution endpoint, so quent's DAG view can scrub through time and show, per pipeline: how many batches (and bytes) are queued waiting for the scheduler, packaged into tasks, or actively processing, and **which memory tier / GPU device the data resides on**. Companion PR (merged): rapidsai/quent#393 — the generic categorical-timeline protocol + playhead UI. ## Model (`rust/crates/telemetry/model`) New `Batch` FSM: `batch_registered{batch_id, pipeline_uuid, port_uuid, origin} → batch_queued → batch_packaged{task_uuid} → batch_processing{task_uuid} → batch_consumed{reason}`, where every non-terminal state carries a tier usage on a new `MemoryTier` resource — **one per GPU device** (`GPU-0`, `GPU-1`, …) plus engine-wide `HOST`/`DISK` — weighted by the batch's bytes. Tier changes (downgrade/spill/prepare-time upgrade) are self-transitions with the new tier usage. Placements share the engine's `batch_id` across fan-out and OOM re-packaging, so batch identity survives task reschedules. Complementary to `DataBatch` (sirius-db#1068), which tracks physical residency; `Batch` tracks scheduling lifecycle per consumer. ## Engine instrumentation (`src/`) A process-global `batch_telemetry_registry` (sharded by batch id) maps port repositories to consumer pipelines (registered during plan telemetry), so publish sites only pass `(batch, repo)`: - Publishes (operator sink, partition consumer) emit `registered → queued` before the batch becomes poppable. - `gpu_pipeline_task`: ctor claims inputs (`queued → packaged`, with lazy registration for OOM-reschedule intermediates), execute emits `packaged → processing` post-prepare (tier re-read to capture upgrades; an id-based path covers merge/concat inputs consumed during materialization), dtor releases claims **by recorded batch id** (the weak batch refs are dead by then). OOM re-claims transfer ownership to the rescheduled task. - Tier moves reported from `convertible_data_batch::convert` and `lock_or_prepare_batch` while the exclusive lock is held (values passed, never re-locked). - Leftovers drained as `consumed{query_end}` in `QueryEnd`; gated by new `telemetry.enable_batch_events` (default on). Every placement in TPC-H SF1/SF300 (2-GPU) test runs completes `consumed{processed}` inside the query window. ## Analyzer (`rust/crates/telemetry/analyzer`) - Ingests Batch/MemoryTier events; implements `UiAnalyzer::data_flow_timeline` (states × tiers × count/bytes per pipeline; `Unsupported` for datasets without batch events, so old recordings keep working and the UI hides the view). - `batch` is a third entity kind in the per-state resource timelines and entity lists, beside `task`/`data_batch`; `batch_registered` (instantaneous bookkeeping entry state) is omitted from aggregated series. - Byte quantities declared with SI prefixes (GB rather than GiB). - 7 analyzer unit tests incl. hand-computed bin math with a mid-queue tier-change split. ##⚠️ Do not merge yet (draft) - The quent crates are pinned to upstream `rapidsai/quent` main (the rapidsai/quent#393 merge commit); the temporary `[patch]` block used during development has been removed. - Includes a merge of latest `dev` (resolved against sirius-db#1068/sirius-db#1080/sirius-db#1112; the sirius-db#1112 per-transition tooltip attributes are wired for batches too). ## Testing - `pixi run make test` (1538 C++ cases) and `cargo test` green; full E2E: TPC-H SF1 + SF300 parquet on 2 GPUs (incl. a constrained-GPU-memory run), NDJSON lifecycle invariants verified, served through the embedded quent UI and scrubbed. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This was referenced Jul 27, 2026
This was referenced Aug 3, 2026
3 tasks
This was referenced Aug 5, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What
Adds a generic data-flow-over-time view to the query-plan DAG: a time slider (playhead) under the DAG that shows, for each operator at time T, how much data is in each lifecycle state (e.g. queued / packaged / processing) and where it resides (an application-defined dimension, e.g. memory tier), animated as you scrub.
quent stays fully application-agnostic: the protocol is per operator, a binned timeline of a distribution over (FSM state × downstream-declared dimension) for downstream-declared measures. All semantics (which FSM, what the dimension keys mean, units) are declared by the downstream analyzer and rendered verbatim.
Backend
crates/analyzer:DistributionTimelineBuilder— span-weighted aggregation over opaque(series, measure, state, dimension)keys, reusing the existingKeyedAggregator.crates/ui:DistributionTimelineRequest/DistributionDecl(dimension keys + measures withQuantitySpecrefs,entity_type_namereferencing the bundle'sfsm_types) /DistributionSeriesDTOs (ts-rs exported).domains/query_engine:UiAnalyzer::data_flow_timelinewith a defaultUnsupportedimpl (existing analyzers compile unchanged; the UI hides the view), andPOST /api/engines/{id}/timeline/data-flow(one request returns all operators of a query; scrubbing is a pure client-side bin lookup). Response shape iscombine_chunks-compatible for future chunked caching.UI
DagPlayhead: plain-DOM scrubber under the DAG (drag, keyboard, play/pause) that also drives a synced crosshair on the resource timelines.NodeFlowBaron every DAG node: a stacked state bar over a dimension (tier) bar, scaled against the fetched window's max so scrubbing shows genuine accumulation; width-gated in-segment value labels (2–3 significant digits + unit, only when they fit); a dual-measure totals line (121 | 5.4GB).Testing
cargo testacross the workspace (new aggregation unit tests + simulator functional tests).Notes for reviewers
docs/domains/query_engine/README.mddocuments the new protocol section.main(clean, no conflicts).🤖 Generated with Claude Code