Skip to content

Commit 4da0bf6

Browse files
committed
T2.9 fix 1: an average's, a minimum's or a maximum's lag on its own row
The reference's engine labels each lag through execute_join, which drops the examples whose window holds no value, and lines the values up by position in its frame of the lags, so an average's, a minimum's or a maximum's lagged values move to the first rows rather than their own, and the last rows (often the entities predicted for) get none. Decided by the orchestrator under the owner's standing delegation (KO2-70, 2026-10-19): each value goes on its own row. - sdm-core's lag::Lag::columns places each label by the labels' mask on the example it belongs to; an example whose window holds none has none. A sum's and a count's lags, every example's, are unchanged. - Tests of my own reading: lag.rs's a_maximum_s_values_keep_their_rows_ across_gaps (a value, a gap, a value, over two steps, rows kept in a request's order, a sum beside it) and the average's test, now on its own row; the route test test_an_average_s_lag_is_each_row_s_own_window (each predicted row's lag its own window's average, read from the data with DuckDB; the rest of the request equal to the old route's), in place of the average's old-route equality. - The 40-source differential again: every payload difference is an average's, a minimum's or a maximum's lag column or a comparison of one. Its register entry and P2B-19's approval cite KO2-70, which is not on the phase head yet; they follow in their own commit once it is (tracker). Signed-off-by: Adi Wadaskar <awadaskar@nvidia.com>
1 parent ae82784 commit 4da0bf6

3 files changed

Lines changed: 131 additions & 28 deletions

File tree

‎docs/design/rust-core/phase-2/tracker-T2.9.md‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -36,10 +36,12 @@ T2.9 is forecasting's options on top of the split already in the core (E153): `n
3636

3737
### Deviations and questions
3838

39-
- **An average, a minimum or a maximum's lag is the reference's, misaligned** (matched, a question for the orchestrator).
39+
- **An average's, a minimum's or a maximum's lag on its own row** (fix 1, decided by the orchestrator under KO2-70 on 2026-10-19).
4040
The reference's executor labels a lag through `execute_join`, which drops the examples whose window holds no value, and the engine's frame of the lags lines the rest up by position, so each value moves to the first examples rather than its own, and the last examples get none (shown: user 0's predicted row at 2023-08-01 gets no lag where its window holds 112.0).
41-
The core matches it (`lag.rs`, its module docs), so the payloads are equal.
42-
Fixing it (each value on its own example) is one line in `Lag::columns` and a register entry; it changes what the model is told, so it is the orchestrator's to decide (KO2-70).
41+
The core puts each value on its own example (`lag.rs`'s `Lag::columns`, by the labels' mask), so these columns differ from the reference's; a sum's, a count's and every comparison of them are unchanged.
42+
Held by `lag.rs`'s `a_maximum_s_values_keep_their_rows_across_gaps` and `an_average_s_value_is_on_its_own_example`, and the route test `test_an_average_s_lag_is_each_row_s_own_window` (each predicted row's lag its own window's average, read from the data; nothing else of the request differs from the old route's).
43+
The 40-source differential after the fix: every payload difference is an average's, a minimum's or a maximum's lag column (or a comparison of one), none elsewhere.
44+
Its register entry waits for an ID (lane b's block, P2B-1 to P2B-19, is spent) and for KO2-70 on the phase head, as does P2B-19's approval.
4345
- **Wording of the refused lags** (the orchestrator's note): the reference refuses `-1` with its engine's own sentence (`'lag_timesteps' cannot be negative (got -1)`), and `1.5`, `2.0` and `true` (`unsupported operand type(s) for *: 'float' and 'DateOffset'`) and `"2"` (`'<' not supported between instances of 'str' and 'int'`) with Python's TypeError text, all `INVALID_REQUEST`.
4446
The core route gives the same code and class, details `field: lag_timesteps` and `reason` (`negative`, `not_a_whole_number`, `not_a_number`), and the Python host says plainly: "lag_timesteps is how many past steps of the target are added as features, a whole number of zero or more; this one is refused (<reason>)" (`_rust/errors.py`'s `WORDED`; test `test_a_refused_lag_is_said_plainly_under_the_reference_s_code`).
4547
A lag past the time range is said as "lag_timesteps reaches further back than a time can be held (before 1677); ask for fewer steps" (P2B-18). The cases hold codes, not messages (H17).

‎packages/structured-data-models-client/tests/test_the_materialize_route.py‎

Lines changed: 50 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -199,7 +199,6 @@ def test_the_regression_s_statistic_is_recorded_as_on_the_old_route(
199199
[
200200
QUERY,
201201
BINARY,
202-
'PREDICT AVG(orders.price, 0, 30, days) FOR EACH users.user_id',
203202
'PREDICT users.age FOR EACH users.user_id',
204203
],
205204
)
@@ -217,6 +216,56 @@ def test_lag_timesteps_builds_the_request_the_old_route_builds(
217216
)
218217

219218

219+
def test_an_average_s_lag_is_each_row_s_own_window(
220+
monkeypatch: pytest.MonkeyPatch,
221+
) -> None:
222+
# T2.9 fix 1 (registered, decided under KO2-70): the reference's engine
223+
# lines an average's lagged values up by position, so they land on the
224+
# first rows rather than their own; the core puts each on its own row.
225+
# Each predicted row's lag is its own window's average, read here from
226+
# the data, and nothing else of the request differs from the old route.
227+
query = 'PREDICT AVG(orders.price, 0, 30, days) FOR EACH users.user_id'
228+
monkeypatch.setenv(_routing.ENVIRONMENT, CORE)
229+
core = _materialize(query=query, lag_timesteps=2)._payloads[0]
230+
monkeypatch.delenv(_routing.ENVIRONMENT)
231+
table = core['predict']['instance_table']
232+
at = table['columns'].index('__kumo_arl0__')
233+
connection = _connection()
234+
for row, user in zip(table['rows'], [0, 1], strict=True):
235+
for k, (start, end) in enumerate(
236+
[('2024-01-31', '2024-03-01'), ('2024-01-01', '2024-01-31')]
237+
):
238+
[(mean,)] = connection.execute(
239+
'SELECT AVG(price) FROM orders WHERE user_id = ? '
240+
'AND ts > ?::TIMESTAMP AND ts <= ?::TIMESTAMP',
241+
[user, start, end],
242+
).fetchall()
243+
got = row[at + k]
244+
if mean is None:
245+
assert got is None
246+
else:
247+
assert got == float(np.float32(mean))
248+
if _old_route_needs_the_engine():
249+
return
250+
old = _materialize(query=query, lag_timesteps=2)._payloads[0]
251+
old = json.loads(json.dumps(old, default=str))
252+
253+
def without_lags(payload: dict[str, Any]) -> dict[str, Any]:
254+
payload = json.loads(json.dumps(payload))
255+
for half in ('context', 'predict'):
256+
table = payload[half]['instance_table']
257+
keep = [
258+
i
259+
for i, c in enumerate(table['columns'])
260+
if not c.startswith('__kumo_arl')
261+
]
262+
table['columns'] = [table['columns'][i] for i in keep]
263+
table['rows'] = [[r[i] for i in keep] for r in table['rows']]
264+
return payload
265+
266+
assert without_lags(core) == without_lags(old)
267+
268+
220269
@pytest.mark.parametrize(
221270
('value', 'reason'),
222271
[

‎rust/crates/sdm-core/src/lag.rs‎

Lines changed: 76 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -23,13 +23,13 @@
2323
//! there (`execute` with `filter_na=False` over a query whose aggregation
2424
//! is joined, so `execute_join` drops what has no value): an aggregation
2525
//! (a sum or a count of every example; an average, a minimum or a maximum
26-
//! of only the examples whose window holds a value) or its comparison, its
27-
//! values in the examples' order, the first value the first example's,
28-
//! and an example past the values none, as the engine's frame of the steps
29-
//! lines them up by position (`pd.DataFrame`, then `pd.concat` with the
30-
//! examples). That is the reference's: an average, a minimum or a maximum
31-
//! moves each value to the first examples rather than its own, which this
32-
//! matches (T2.9's tracker).
26+
//! of only the examples whose window holds a value) or its comparison, each
27+
//! value on its own example and an example whose window holds none
28+
//! without one. The reference's engine lines the values up by position
29+
//! instead (`pd.DataFrame`, then `pd.concat` with the examples), so an
30+
//! average's, a minimum's or a maximum's move to the first examples rather
31+
//! than their own; the core puts each on its own row (T2.9 fix 1, a
32+
//! registered difference decided under KO2-70).
3333
3434
use sdm_contract::{CoreError, Detail, ErrorCode, ErrorKind, Scalar};
3535
use sdm_engine::labels::{self, Aggregation, Constant, Labels, RelOp, Target};
@@ -190,15 +190,25 @@ impl Lag {
190190
constant,
191191
},
192192
};
193-
// The values of the examples that have one, in order, each
194-
// at the position of the first examples (see the module).
195-
let values = match labels::label(&target)?.labels {
196-
Labels::Float(v) => {
197-
LagValues::Float(kept.iter().map(|&at| v.get(at).copied()).collect())
198-
}
199-
Labels::Bool(v) => {
200-
LagValues::Bool(kept.iter().map(|&at| v.get(at).copied()).collect())
201-
}
193+
// Each value on its own example: the labels are those of the
194+
// examples `mask` marks, in order (see the module).
195+
let labelled = labels::label(&target)?;
196+
let mut at_example = vec![None; examples];
197+
for (slot, example) in (0..examples).filter(|&e| labelled.mask[e]).enumerate() {
198+
at_example[example] = Some(slot);
199+
}
200+
let own = |at: usize| at_example.get(at).copied().flatten();
201+
let values = match labelled.labels {
202+
Labels::Float(v) => LagValues::Float(
203+
kept.iter()
204+
.map(|&at| own(at).and_then(|i| v.get(i).copied()))
205+
.collect(),
206+
),
207+
Labels::Bool(v) => LagValues::Bool(
208+
kept.iter()
209+
.map(|&at| own(at).and_then(|i| v.get(i).copied()))
210+
.collect(),
211+
),
202212
Labels::Int(_) | Labels::Text(_) => {
203213
return Err(CoreError::new(
204214
ErrorKind::Internal,
@@ -264,11 +274,12 @@ mod tests {
264274
);
265275
}
266276

267-
/// An average has a value only where its window holds one, and the
268-
/// values line up with the first examples (the reference's frame);
269-
/// a comparison of it likewise.
277+
/// An average has a value only where its window holds one, each on
278+
/// its own example, the first example's none (the reference's frame
279+
/// would move the second's value to it; T2.9 fix 1); a comparison of
280+
/// it likewise.
270281
#[test]
271-
fn an_average_s_values_line_up_with_the_first_examples() {
282+
fn an_average_s_value_is_on_its_own_example() {
272283
let w = window(1, days(0), days(10), &[100 * DAY, 50 * DAY]).unwrap();
273284
let lag = |condition| Lag {
274285
steps: 1,
@@ -286,18 +297,59 @@ mod tests {
286297
};
287298
assert_eq!(
288299
lag(None).columns(&[0, 1]).unwrap()[0].values,
289-
LagValues::Float(vec![Some(3.0), None])
300+
LagValues::Float(vec![None, Some(3.0)])
290301
);
291302
assert_eq!(
292303
lag(Some((RelOp::Gt, Some(Constant::Int(1)))))
293304
.columns(&[0, 1])
294305
.unwrap()[0]
295306
.values,
296-
LagValues::Bool(vec![Some(true), None])
307+
LagValues::Bool(vec![None, Some(true)])
308+
);
309+
assert_eq!(
310+
lag(Some((RelOp::Eq, None))).columns(&[1, 0]).unwrap()[0].values,
311+
LagValues::Bool(vec![Some(false), None])
312+
);
313+
}
314+
315+
/// Examples whose windows hold a value, none, then a value, over two
316+
/// steps (T2.9 fix 1): a maximum's each on its own row, a gap where a
317+
/// window is empty, for the context and the entities predicted for
318+
/// alike; a sum's every row.
319+
#[test]
320+
fn a_maximum_s_values_keep_their_rows_across_gaps() {
321+
let anchors = [100 * DAY, 100 * DAY, 100 * DAY];
322+
let w = window(2, days(0), days(10), &anchors).unwrap();
323+
// Example 0: rows at days 85 (5.0) and 95 (7.0); example 1: none;
324+
// example 2: a row at day 88 (2.0) only.
325+
let rows = || Rows {
326+
batches: vec![0, 0, 2],
327+
values: labels::Values::Float(vec![Some(5.0), Some(7.0), Some(2.0)]),
328+
times: vec![Some(85 * DAY), Some(95 * DAY), Some(88 * DAY)],
329+
};
330+
let lag = |aggregation| Lag {
331+
steps: 2,
332+
aggregation,
333+
start: days(0),
334+
end: days(10),
335+
condition: None,
336+
shifted: w.shifted.clone(),
337+
rows: rows(),
338+
};
339+
let max = lag(Aggregation::Max).columns(&[0, 1, 2]).unwrap();
340+
// __kumo_arl0__ is (90, 100], __kumo_arl1__ (80, 90].
341+
assert_eq!(max[0].values, LagValues::Float(vec![Some(7.0), None, None]));
342+
assert_eq!(
343+
max[1].values,
344+
LagValues::Float(vec![Some(5.0), None, Some(2.0)])
297345
);
346+
// The rows a request keeps, in their own order.
347+
let kept = lag(Aggregation::Max).columns(&[2, 0]).unwrap();
348+
assert_eq!(kept[1].values, LagValues::Float(vec![Some(2.0), Some(5.0)]));
349+
let sum = lag(Aggregation::Sum).columns(&[0, 1, 2]).unwrap();
298350
assert_eq!(
299-
lag(Some((RelOp::Eq, None))).columns(&[1]).unwrap()[0].values,
300-
LagValues::Bool(vec![None])
351+
sum[1].values,
352+
LagValues::Float(vec![Some(5.0), Some(0.0), Some(2.0)])
301353
);
302354
}
303355

0 commit comments

Comments
 (0)