From 5cf9831aff60209d70e3d5d8f3b2c85ae248b47f Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sat, 31 Oct 2026 17:29:12 -0400 Subject: [PATCH] DDIR reduce: integer values are their own ids A reduce's value ids named rows of a pool, numbered per window by sorting each key's values with corgi's segmented sort. A value that is a primitive integer (perhaps in one-field products) now names itself: its bits with the sign bit flipped, so ids ascend in value order and equal values share one without a sort. The reduction reads such values from their ids, corrections are a consolidation of ids, and emission writes the ids back as the value column. The input decides by the values `present` compares: their projection, where a weighted sum is proven to merge equal projections, else the whole value. A pair read at one integer field projects to `(int)`, and names itself. Its integers are read from the projected columns, where they are held as `i64`s, else gathered. The output decides by the reduction's output value shape. Both decide per window: only where each identifier holds one real key, always for integer keys and for hashed keys unless the window has a collision, where two real keys under one identifier could hold equal values. Such windows keep pool ids, and a batch built from both kinds of window gathers from the pool and the values together. `present` now builds each window's labels and rows at their final size. Grown by doubling, they cost a presentation up to half again its time, depending on the allocator's state: a projected pair sum ran its presentations at 3.1 or 2.1 ms, and its retires at 6.4 or 2.6 ms, by the launch. Reserved, it runs at 2.0 and 2.5. Implemented and measured by a Claude subagent. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_019XsakwFGwtESwZ1gSYwG4o --- interactive/src/runtime/reduce.rs | 476 +++++++++++++++++++++++++++--- 1 file changed, 429 insertions(+), 47 deletions(-) diff --git a/interactive/src/runtime/reduce.rs b/interactive/src/runtime/reduce.rs index d5c666d6a..f24d1fafe 100644 --- a/interactive/src/runtime/reduce.rs +++ b/interactive/src/runtime/reduce.rs @@ -9,11 +9,20 @@ //! key's equal input values share one id, found by corgi's segmented sort. A proven weighted //! sum compares projected values here; other reducers keep whole-value identity. Output values are //! matched to the current output by structural comparison. Nothing hashes values. +//! * integer values — a value that is a primitive integer (perhaps in one-field products) is its +//! own id, mapped into `u64` with its sign bit flipped so that ids ascend in value order. Equal +//! values share an id with no sort, and no pool row names them: the reduction reads its values +//! from the ids, and emission writes the ids back as the value column. Input and output decide +//! this apart, by the shapes of the values they compare (for input, the projected values where +//! a weighted sum compares those, so a wide value read at one integer field counts), in each +//! window whose identifiers each hold one real key: always for integer keys, and for hashed +//! keys unless the window has a collision. //! * collisions — a hashed key's identifier may hold several real keys. Chunks sort each //! identifier's rows by real key, so a window finds them with two comparisons per run. Their //! ids are then per real key, input and output alike: a colliding bracket is split by real key //! before it is reduced, its output is matched by key as well as value, and every output id -//! records the row holding its key, which is what `emit` reads. +//! records the row holding its key, which is what `emit` reads. A value is not its own id in +//! such a window, as two real keys under one identifier may hold equal values. //! * the value callback — `reduce_lambda` runs ONE crossing per wave over every `(key, time)` //! bracket, evaluating its expression over every bracket's key and `(val, diff)` list. //! The resulting `(val, diff)` lists supply the desired output columns, never DDIR rows. @@ -22,28 +31,30 @@ //! * materialize — gather the emitted keys and values from the pools' columns and seal a //! `CorgiChunk` batch column-natively. //! -//! Presentation assigns value ids in structural order (signed integers, lexicographic lists). -//! Reduction expressions read their input lists in that order. +//! Presentation assigns value ids in structural order (signed integers, lexicographic lists), +//! whether ids name rows or are the values themselves. Reduction expressions read their input +//! lists in that order. //! //! The changed-key restriction is honored by presenting only the changed keys: novel batches are //! read whole (delta-sized), the accumulated history is scanned and filtered to the changed hashes //! (a columnar semijoin — matching the row-wise tactic's read). +use std::borrow::Cow; use std::ops::Range; use std::rc::Rc; -use differential_dataflow::consolidation::consolidate; +use differential_dataflow::consolidation::{consolidate, consolidate_from}; use differential_dataflow::trace::Description; use differential_dataflow::trace::chunk::ChunkBatch; use differential_dataflow::operators::int_proxy::{KeyPosition, ProxyBridge}; use differential_dataflow::operators::int_proxy::reduce::{ProxyReduceBackend, ReduceInstance, ReduceWindow}; -use corgi::arrange::{compare_at, equal_idx, gather_lanes, sort_blocks}; +use corgi::arrange::{compare_at, equal_idx, gather_lanes, leaf_slice, sort_blocks}; use corgi::{Bounds, Demand, Graph, NumOp, Shape, Value as CValue}; use crate::runtime::col_times::{ColTime, ColTimes}; use crate::runtime::search::matching_ranges; -use crate::runtime::chunk::{colliding_keys, columns_to_batch, ident, key_ids, key_is_hashed, recover_key, recover_key_ref, CorgiChunk}; +use crate::runtime::chunk::{colliding_keys, columns_to_batch, ident, int_lane, key_ids, key_is_hashed, recover_key, recover_key_ref, CorgiChunk}; use crate::lambda::compile_reduce; use crate::ir::{Diff, Reducer}; @@ -70,6 +81,41 @@ fn gather_refs(columns: &[CValue], refs: impl Iterator) - gather_lanes(&srcs, &tags, &offs) } +/// How many one-field products wrap an integer leaf of this shape, if it is one. Such a value +/// orders as its integer does, and can be its own id. +fn int_leaf(shape: &Shape) -> Option { + match shape { + Shape::Int => Some(0), + Shape::Prod(fields) if fields.len() == 1 => int_leaf(&fields[0]).map(|depth| depth + 1), + _ => None, + } +} + +/// An integer's id: its bits with the sign bit flipped, so that ids ascend as integers do. +fn int_id(value: i64) -> u64 { (value as u64) ^ (1 << 63) } + +/// The integer an id names: the inverse of [`int_id`]. +fn id_int(id: u64) -> i64 { (id ^ (1 << 63)) as i64 } + +/// The integers `ids` name, as a column in `depth` one-field products. +fn int_column(depth: usize, ids: impl Iterator) -> CValue { + let mut column = CValue::i64(ids.map(id_int).collect()); + for _ in 0..depth { column = CValue::Prod(vec![column]); } + column +} + +/// The ids of the integers at `rows`, each a `(column, row)` of `columns`. Columns held as `i64`s, +/// as chunks hold integer values, are read where they are. If a row's column is held narrower, +/// as a projected field may be, the rows are gathered and read together instead. +fn int_ids(columns: &[CValue], rows: &[(usize, usize)]) -> Vec { + let lanes: Vec> = columns.iter().map(int_lane).collect(); + if rows.iter().all(|&(column, _)| lanes[column].is_some()) { + return rows.iter().map(|&(column, row)| int_id(lanes[column].unwrap()[row])).collect(); + } + let gathered = gather_refs(columns, rows.iter().copied()); + leaf_slice(&gathered).expect("integer values").iter().map(|&value| int_id(value)).collect() +} + /// Where each of a list of chunks holds some ascending keys: per chunk, the `(key index, rows)` of /// each key it holds, in key order. type Matches = Vec)>>; @@ -154,12 +200,34 @@ pub(crate) struct CorgiReduceBackend { key_columns: Vec, /// The output emitted so far: key rows in `key_columns`, value ids, times, and diffs. rows: (Vec<(usize, usize)>, Vec, ColTimes, Vec), + /// The emitted rows whose ids are output values rather than rows of `output`, as ranges. + value_rows: Vec>, + /// If the input's values are integers, in how many one-field products. + int_input: Option, + /// If the output's values are integers, in how many one-field products. + int_output: Option, + /// How the current window's ids name values. + window_ids: WindowIds, /// Input records per window, bounding what the presentations cost at once. window_size: usize, /// The retire in progress. retire: Retire, } +/// How the current window's ids name values: by rows of the pools, or as the integers themselves. +/// Integers name themselves where the values are integers and each of the window's identifiers +/// holds one real key. +#[derive(Default)] +struct WindowIds { + /// Whether input ids are the integer values themselves. + input_values: bool, + /// Whether output ids are the integer values themselves. + output_values: bool, + /// When either is, a row of `key_columns` holding each of the window's keys, if any does, as + /// such ids record none. + key_rows: Vec>, +} + /// A retire's keys and where its input chunks hold them, found once by its first window. #[derive(Default)] struct Retire { @@ -215,6 +283,7 @@ impl CorgiReduceBackend { let Reducer::Expr(lambda) = reducer; let graph = compile_reduce(&lambda, &kshape, &vshape) .unwrap_or_else(|e| panic!("reduce: type error at shapes ({kshape}, {vshape}): {e}")); + let value_shape = vshape.clone(); // The graph rebuilt to read only what it reads, and the demand saying what that is. let input = Shape::Prod(vec![kshape, Shape::List(Box::new(Shape::Prod(vec![vshape, Shape::Int])))]); let compiled = corgi::compile(graph, &input) @@ -224,9 +293,24 @@ impl CorgiReduceBackend { let project_values = limit.is_none() && crate::passes::reduce_demand::projection(&compiled).is_some(); let narrowed = corgi::narrow(compiled, &Demand::All) .unwrap_or_else(|e| panic!("reduce: narrowing a compiled graph at {input}: {e}")); + let reads = Reads::new(narrowed.input, &input); + // Input ids name the values `present` compares: projected, where the reducer merges equal + // projections (see `projects`), else whole. + let compared = if project_values && reads.value != Demand::All { + reads.value.projected_shape(&value_shape).expect("a demand on the reduce's values") + } else { value_shape }; + let int_input = int_leaf(&compared); + // The graph's output is a list of `(val, diff)`. + let int_output = match corgi::shape_of(&narrowed.graph, &reads.shape) { + Ok(Shape::List(element)) => match *element { + Shape::Prod(fields) if fields.len() == 2 => int_leaf(&fields[0]), + _ => None, + }, + _ => None, + }; CorgiReduceBackend { graph: narrowed.graph, - reads: Reads::new(narrowed.input, &input), + reads, project_values, limit, input: Pool::default(), @@ -234,6 +318,10 @@ impl CorgiReduceBackend { output_keys: Vec::new(), key_columns: Vec::new(), rows: (Vec::new(), Vec::new(), ColTimes::default(), Vec::new()), + value_rows: Vec::new(), + int_input, + int_output, + window_ids: WindowIds::default(), window_size: window_size.max(1), retire: Retire::default(), } @@ -254,21 +342,32 @@ fn search(chunks: &[&CorgiChunk], keys: &[i64]) -> Matches }).collect() } +/// How [`present`] names each run of equal values under a key. +enum Naming<'a> { + /// By a row: the next index of `refs`, where the run is recorded as the `(chunk, row)` of one + /// of its rows. + Rows(&'a mut Vec<(usize, usize)>), + /// By the value itself, an integer. Exact only where each identifier holds one real key. + Ints, +} + /// Present the records `chunks` hold for `keys`, at `matches`, into `bridge`, sorted by /// `((key, id), time)` and consolidated. /// -/// Each key's rows are sorted by value, and each run of equal values gets one id: the next index -/// of `refs`, where it is recorded as the `(chunk, row)` of one of its rows. The keys `colliding` -/// marks (by index; empty when none do) hold several real keys under one identifier: their rows are -/// first split by real key, so that an id names one real key's value, and the row it is recorded -/// at holds that key. +/// Values are compared as `projected` holds them, a column per chunk, or else whole. Each key's +/// rows are ordered by value, and each run of equal values gets one id, as `naming` says. +/// Integers are ordered by their ids, which a chunk holds in order, so only a key that several +/// chunks hold is sorted, and as integers. Otherwise rows are sorted by value. The keys +/// `colliding` marks (by index; empty when none do) hold several real keys under one identifier: +/// their rows are first split by real key, so that an id names one real key's value, and the row +/// it is recorded at holds that key. Their values are never named by themselves. fn present( chunks: &[&CorgiChunk], keys: &[i64], matches: &Matches, colliding: &[bool], projected: Option<&[CValue]>, - refs: &mut Vec<(usize, usize)>, + naming: Naming<'_>, bridge: &mut ProxyBridge, ) { // The matched `(chunk, row)`s, key by key, and each one's key index. @@ -276,7 +375,10 @@ fn present( .flat_map(|(chunk, found)| found.iter().map(move |(key, rows)| (*key, chunk, rows.clone()))) .collect(); runs.sort_by_key(|run| run.0); - let (mut labels, mut rows) = (Vec::new(), Vec::new()); + // At their final size: grown by doubling, window after window, these cost a presentation up to + // half again its time, by the allocator's state. + let total = runs.iter().map(|run| run.2.len()).sum(); + let (mut labels, mut rows) = (Vec::with_capacity(total), Vec::with_capacity(total)); for (key, chunk, range) in runs { labels.extend(range.clone().map(|_| key as u64)); rows.extend(range.map(|row| (chunk, row))); @@ -284,21 +386,40 @@ fn present( if rows.is_empty() { return; } - // Each label's key index; the identity unless real keys split some labels. - let mut label_keys = None; - if colliding.contains(&true) { - (labels, rows, label_keys) = split_real_keys(chunks, colliding, labels, rows); - } - let key_of = |label: u64| label_keys.as_ref().map_or(label as usize, |keys: &Vec| keys[label as usize]); // Output and unproven input reducers retain whole-value identity and order. let whole; let vals = if let Some(projected) = projected { projected } else { whole = chunks.iter().map(|chunk| chunk.vals().clone()).collect::>(); &whole }; - let (sorted, groups) = sort_blocks(&labels, &gather_refs(vals, rows.iter().copied())); let update = |(chunk, row): (usize, usize)| (chunks[chunk].times().get(row), chunks[chunk].diffs()[row]); - let (mut start, mut updates) = (0, Vec::new()); + let mut updates = Vec::new(); + let refs = match naming { + Naming::Rows(refs) => refs, + Naming::Ints => { + assert!(!colliding.contains(&true), "present: a value names itself only under an identifier holding one real key"); + let ids = int_ids(vals, &rows); + let mut order: Vec = (0..rows.len()).collect(); + for block in order.chunk_by_mut(|&a, &b| labels[a] == labels[b]) { + if !block.is_sorted_by_key(|&i| ids[i]) { block.sort_by_key(|&i| ids[i]); } + } + for group in order.chunk_by(|&a, &b| labels[a] == labels[b] && ids[a] == ids[b]) { + updates.extend(group.iter().map(|&i| update(rows[i]))); + consolidate(&mut updates); + let (key, id) = (keys[labels[group[0]] as usize], ids[group[0]]); + bridge.extend(updates.drain(..).map(|(time, diff)| ((key, id), time, diff))); + } + return; + } + }; + // Each label's key index; the identity unless real keys split some labels. + let mut label_keys = None; + if colliding.contains(&true) { + (labels, rows, label_keys) = split_real_keys(chunks, colliding, labels, rows); + } + let key_of = |label: u64| label_keys.as_ref().map_or(label as usize, |keys: &Vec| keys[label as usize]); + let (sorted, groups) = sort_blocks(&labels, &gather_refs(vals, rows.iter().copied())); + let mut start = 0; for group in groups.chunk_by(|a, b| a == b) { let members = &sorted[start..start + group.len()]; start += group.len(); @@ -370,8 +491,8 @@ where /// The harness presents each bracket consolidated, so its diffs are non-zero and its value ids /// ascend, which is value order: `present` numbers each key's values in sorted order. The /// brackets are presented as they are. One with no input has no output, and the lambda does not - /// see it. - fn reduce_lambda(&self, ends: &[usize], input: &[(u64, Diff)]) -> (CValue, Vec, Vec) { + /// see it. `key_rows` holds the row holding each bracket's key, for those with input. + fn reduce_lambda(&self, ends: &[usize], input: &[(u64, Diff)], key_rows: &[Option<(usize, usize)>]) -> (CValue, Vec, Vec) { // A limit hands the lambda each bracket's first values only; nothing past them is gathered. let limited = self.limit.and_then(|k| first_entries(k, ends, input)); let (ends, input) = limited.as_ref().map_or((ends, input), |(ends, input)| (&ends[..], &input[..])); @@ -381,7 +502,7 @@ where for (bi, &end) in ends.iter().enumerate() { if end > start { blocks.push((bi, end)); - key_refs.push(self.input.refs[input[start].0 as usize]); + key_refs.push(key_rows[bi].expect("a bracket with input has a row holding its key")); } start = end; } @@ -390,7 +511,7 @@ where } let vals = match self.reads.value { Demand::None => CValue::Unit(input.len()), - _ => self.input.gather(&input.iter().map(|&(id, _)| id).collect::>()), + _ => self.input_values(input.iter().map(|&(id, _)| id)), }; let diffs = CValue::i64(input.iter().map(|&(_, diff)| diff).collect()); // Projected values may tie; whole ones ascend. @@ -423,6 +544,73 @@ where } (out_vals, out_diffs, out_ends) } + + /// The current window's input values that `ids` name, projected by what the graph reads of + /// them. + fn input_values(&self, ids: impl Iterator) -> CValue { + match self.int_input.filter(|_| self.window_ids.input_values) { + // Ids of projected values are already what the graph reads. + Some(depth) if self.projects() => int_column(depth, ids), + Some(depth) => self.reads.value.project(int_column(depth, ids)).expect("an integer has the reduce's value shape"), + None => self.input.gather(&ids.collect::>()), + } + } + + /// Whether input values are compared as the graph reads them: the reducer is proven to merge + /// equal projected values, and it does not read them whole. + fn projects(&self) -> bool { self.project_values && self.reads.value != Demand::All } + + /// The row of `key_columns` holding each bracket's key, for the brackets with input. Ids that + /// name rows of the input pool name rows holding their keys. Ids that are values do not, and + /// then each bracket is a key of the window (no identifier collides), whose row it has. + fn bracket_rows(&self, keys: &[i64], ends: &[usize], input: &[(u64, Diff)]) -> Vec> { + let window = &self.retire.keys[self.retire.window.clone()]; + debug_assert!(!self.window_ids.input_values || keys.len() == ends.len(), "reduce: a bracket per key"); + let mut start = 0; + ends.iter().enumerate().map(|(bracket, &end)| { + let first = std::mem::replace(&mut start, end); + (end > first).then(|| if self.window_ids.input_values { + let at = window.binary_search(&keys[bracket]).expect("a bracket's key is in its window"); + self.window_ids.key_rows[at].expect("a key with input has a row holding it") + } else { + self.input.refs[input[first].0 as usize] + }) + }).collect() + } + + /// [`ProxyReduceBackend::reduce_corrections`] where output ids are values: each key's desired + /// rows, as ids, less its output, consolidated. Brackets are keys, as no identifier collides. + fn value_corrections(&self, keys: &[i64], in_ends: &[usize], input: &[(u64, Diff)], out_ends: &[usize], output: &[(u64, Diff)]) -> (Vec<(u64, Diff)>, Vec) { + let key_rows = self.bracket_rows(keys, in_ends, input); + let (column, diffs, ends) = self.reduce_lambda(in_ends, input, &key_rows); + let desired = if diffs.is_empty() { Cow::Borrowed(&[][..]) } else { leaf_slice(&column).expect("the reduce's output values are integers") }; + let (mut corr, mut corr_ends) = (Vec::with_capacity(output.len() + diffs.len()), Vec::with_capacity(keys.len())); + let (mut ds, mut os) = (0, 0); + for (&de, &oe) in ends.iter().zip(out_ends) { + let start = corr.len(); + corr.extend(output[os..oe].iter().map(|&(id, diff)| (id, -diff))); + corr.extend(desired[ds..de].iter().zip(&diffs[ds..de]).map(|(&value, &diff)| (int_id(value), diff))); + consolidate_from(&mut corr, start); + corr_ends.push(corr.len()); + (ds, os) = (de, oe); + } + (corr, corr_ends) + } + + /// The output values the emitted `ids` name: the ids themselves in `value_rows`, and rows of + /// the output pool elsewhere. + fn output_values(&self, ids: &[u64], value_rows: &[Range]) -> CValue { + let Some(depth) = self.int_output.filter(|_| !value_rows.is_empty()) else { return self.output.gather(ids) }; + let values = int_column(depth, ids.iter().copied()); + if value_rows.len() == 1 && value_rows[0] == (0..ids.len()) { return values; } + // Some windows named rows of the pool (an identifier collided): gather from both. + let mut named = vec![false; ids.len()]; + for rows in value_rows { named[rows.clone()].fill(true); } + let mut columns = self.output.columns.clone(); + columns.push(values); + let at = columns.len() - 1; + gather_refs(&columns, ids.iter().zip(named).enumerate().map(|(row, (&id, named))| if named { (at, row) } else { self.output.refs[id as usize] })) + } } impl ProxyReduceBackend, CBatch> for CorgiReduceBackend @@ -437,6 +625,7 @@ where fn begin(&mut self, _description: Description) { self.rows = (Vec::new(), Vec::new(), ColTimes::default(), Vec::new()); + self.value_rows.clear(); } fn next_window(&mut self, instance: &ReduceInstance<'_, T, CBatch, CBatch>, changed: &[i64], from: &mut KeyPosition, window: &mut ReduceWindow) { @@ -465,6 +654,7 @@ where } // Windows are bounded, as their presentations are all live at once. + let projects = self.projects(); let retire = &mut self.retire; if !retire.advance(self.window_size) { *from = KeyPosition::End; @@ -484,6 +674,19 @@ where retire.colliding = keys.iter().zip(&colliding).filter(|(_, c)| **c).map(|(key, _)| *key).collect(); let colliding = if retire.colliding.is_empty() { &[][..] } else { &colliding[..] }; + // Integer values name themselves where each identifier holds one real key. Their ids then + // record no rows, so the window notes a row holding each key, input or output. + let ids = &mut self.window_ids; + ids.input_values = colliding.is_empty() && self.int_input.is_some(); + ids.output_values = colliding.is_empty() && self.int_output.is_some(); + ids.key_rows.clear(); + if ids.input_values || ids.output_values { + ids.key_rows.resize(keys.len(), None); + for (column, found) in matches.iter().chain(&out_matches).enumerate() { + for (key, rows) in found { ids.key_rows[*key].get_or_insert((column, rows.start)); } + } + } + // The seeds are the novel batches' RAW (key, time) support, recorded here — before the // merged presentation below, whose consolidation may net a novel record away entirely. window.seeds.reserve(matches[prior..].iter().flatten().map(|(_, rows)| rows.len()).sum()); @@ -499,13 +702,18 @@ where // equal values share an id, so an exactly cancelling pair vanishes here, and its time // survives in `window.seeds` above. Input ids name rows for this window only. self.input.refs.clear(); - let projected = (self.project_values && self.reads.value != Demand::All).then_some(&self.input.columns[..]); - present(&in_chunks, keys, &matches, colliding, projected, &mut self.input.refs, &mut window.input); + let projected = projects.then_some(&self.input.columns[..]); + let naming = if ids.input_values { Naming::Ints } else { Naming::Rows(&mut self.input.refs) }; + present(&in_chunks, keys, &matches, colliding, projected, naming, &mut window.input); // The output history, same keys. Output ids name rows for the whole retire, and the rows // that hold their keys are at the same positions of the output chunks' key columns. - let fresh = self.output.refs.len(); - present(&out_chunks, keys, &out_matches, colliding, None, &mut self.output.refs, &mut window.output); - self.output_keys.extend(self.output.refs[fresh..].iter().map(|&(chunk, row)| (in_chunks.len() + chunk, row))); + if ids.output_values { + present(&out_chunks, keys, &out_matches, colliding, None, Naming::Ints, &mut window.output); + } else { + let fresh = self.output.refs.len(); + present(&out_chunks, keys, &out_matches, colliding, None, Naming::Rows(&mut self.output.refs), &mut window.output); + self.output_keys.extend(self.output.refs[fresh..].iter().map(|&(chunk, row)| (in_chunks.len() + chunk, row))); + } *from = retire.keys.get(retire.window.end).map_or(KeyPosition::End, |key| KeyPosition::At(*key)); } @@ -520,6 +728,12 @@ where // A colliding identifier's bracket holds several real keys' values. It is split into one // bracket per real key, each reduced on its own, and its rows are matched against output // by key as well as value. Its input ids ascend by real key, so each key's are contiguous. + // + // Where output ids are the values themselves, equal values are equal ids, and the + // correction is a consolidation. + if self.window_ids.output_values { + return self.value_corrections(keys, in_ends, input, out_ends, output); + } let split = keys.iter().any(|key| self.retire.colliding.binary_search(key).is_ok()); let (mut sub_ends, mut sub_input, mut bracket_ends) = (Vec::new(), Vec::new(), Vec::new()); if split { @@ -541,7 +755,8 @@ where } let (in_ends, input) = if split { (&sub_ends[..], &sub_input[..]) } else { (in_ends, input) }; // The desired rows, and each one's diff where it is not one copy. - let (desired, desired_diffs, desired_ends) = self.reduce_lambda(in_ends, input); + let key_rows = self.bracket_rows(keys, in_ends, input); + let (desired, desired_diffs, desired_ends) = self.reduce_lambda(in_ends, input, &key_rows); let fresh = self.output.columns.len(); let prior_refs = self.output.refs.len(); if !desired.is_empty() { self.output.columns.push(desired); } @@ -601,8 +816,7 @@ where let (lo, hi) = (if bs == 0 { 0 } else { desired_ends[bs - 1] }, desired_ends[be - 1]); let rows = (oe - os) + (hi - lo); if !checked && !resolved && hi - lo > 16 && hi - lo > 2 * (usize::BITS - rows.leading_zeros()) as usize { - let first = if bs == 0 { 0 } else { in_ends[bs - 1] }; - let (kc, kr) = self.input.refs[input[first].0 as usize]; + let (kc, kr) = key_rows[bs].expect("a bracket with desired rows has input"); // (column, row, desired, position): of equal values, existing rows come first, // each side in its own order, so a run's first item is the search's answer. order.clear(); @@ -644,8 +858,7 @@ where let (ds, de) = (if s == 0 { 0 } else { desired_ends[s - 1] }, desired_ends[s]); if ds == de { continue; } // A desired row's key: that of the input it was reduced from, which is non-empty. - let first = if s == 0 { 0 } else { in_ends[s - 1] }; - let (kc, kr) = self.input.refs[input[first].0 as usize]; + let (kc, kr) = key_rows[s].expect("a bracket with desired rows has input"); for row in ds..de { let diff = desired_diffs[row]; let equal = |&(id, _): &(u64, Diff)| { @@ -689,21 +902,42 @@ where } fn emit(&mut self, records: &[((i64, u64), T, Diff)]) { - // An output id names its key's row, as two keys may share an identifier. let (key_rows, ids, times, diffs) = &mut self.rows; - for ((_, id), time, diff) in records { - key_rows.push(self.output_keys[*id as usize]); - ids.push(*id); - times.push(time); - diffs.push(*diff); + let start = ids.len(); + if self.window_ids.output_values { + // An output id is the value, and the window holds a row of each key; records ascend + // by key, so the window's keys are walked once. + let window = &self.retire.keys[self.retire.window.clone()]; + let mut at = 0; + for ((key, id), time, diff) in records { + if window.get(at) != Some(key) { at = window.binary_search(key).expect("an emitted key is in its window"); } + key_rows.push(self.window_ids.key_rows[at].expect("an emitted key has a row holding it")); + ids.push(*id); + times.push(time); + diffs.push(*diff); + } + match self.value_rows.last_mut() { + Some(rows) if rows.end == start => rows.end = ids.len(), + _ if ids.len() > start => self.value_rows.push(start..ids.len()), + _ => {} + } + } else { + // An output id names its key's row, as two keys may share an identifier. + for ((_, id), time, diff) in records { + key_rows.push(self.output_keys[*id as usize]); + ids.push(*id); + times.push(time); + diffs.push(*diff); + } } } fn finish(&mut self) -> Option> { let (key_rows, ids, times, diffs) = std::mem::take(&mut self.rows); + let value_rows = std::mem::take(&mut self.value_rows); if times.is_empty() { return None; } let keys = gather_refs(&self.key_columns, key_rows.into_iter()); - Some(Rc::new(columns_to_batch(keys, self.output.gather(&ids), times, diffs))) + Some(Rc::new(columns_to_batch(keys, self.output_values(&ids, &value_rows), times, diffs))) } } @@ -807,7 +1041,7 @@ mod tests { let owned = chunks(&rows, size, key_shape, val_shape); let chunks: Vec<_> = owned.iter().collect(); let (mut refs, mut bridge) = (Vec::new(), Vec::new()); - present(&chunks, &keys, &search(&chunks, &keys), &[], None, &mut refs, &mut bridge); + present(&chunks, &keys, &search(&chunks, &keys), &[], None, Naming::Rows(&mut refs), &mut bridge); assert!(bridge.windows(2).all(|w| (w[0].0, &w[0].1) < (w[1].0, &w[1].1))); let pool = Pool { columns: chunks.iter().map(|chunk| chunk.vals().clone()).collect(), refs }; let values = rows_of(&pool.gather(&bridge.iter().map(|r| r.0.1).collect::>())); @@ -821,6 +1055,46 @@ mod tests { } } + /// Integers named by themselves: `present` gives what the chunks hold, sorted and consolidated, + /// with each value's own id, so a key's ids ascend in signed order, across key shapes, a + /// one-field product, a pair projected to a field held as bytes, chunkings, ties, and + /// cancellations. + #[test] + fn present_names_integers_by_themselves() { + const SIGNED: [i64; 3] = [i64::MIN, -1, 256]; + let rows = random_rows(0x9E37_79B9_7F4A_7C15, 300, 6); + // Each shape, and whether values are compared at their first field. + let val_shapes: [(fn(&[(u64, u64)]) -> CValue, bool); 3] = [ + (|v| CValue::i64(v.iter().map(|v| SIGNED[v.0 as usize]).collect()), false), + (|v| CValue::Prod(vec![CValue::int(v.iter().map(|v| v.1 as i64))]), false), + (|v| CValue::Prod(vec![CValue::int(v.iter().map(|v| v.0 as i64)), CValue::int(v.iter().map(|v| v.1 as i64))]), true), + ]; + let first = Demand::Prod(vec![Demand::All, Demand::None]); + for (key_shape, (val_shape, projected)) in KEY_SHAPES.iter().flat_map(|k| val_shapes.iter().map(move |v| (*k, *v))) { + let mut keys = key_ids(&key_shape(&[0, 2, 3, 5, 9])); + keys.sort(); + for size in [1, 7, rows.len()] { + let owned = chunks(&rows, size, key_shape, val_shape); + let chunks: Vec<_> = owned.iter().collect(); + let columns: Vec = if projected { chunks.iter().map(|chunk| first.project(chunk.vals().clone()).unwrap()).collect() } else { Vec::new() }; + let mut bridge = Vec::new(); + present(&chunks, &keys, &search(&chunks, &keys), &[], projected.then_some(&columns[..]), Naming::Ints, &mut bridge); + assert!(bridge.windows(2).all(|w| (w[0].0, &w[0].1) < (w[1].0, &w[1].1))); + assert!(bridge.windows(2).all(|w| w[0].0.0 != w[1].0.0 || id_int(w[0].0.1) <= id_int(w[1].0.1)), "ids ascend in signed order"); + let values = rows_of(&int_column(0, bridge.iter().map(|r| r.0.1))); + let actual: BTreeMap<_, _> = bridge.iter().zip(values).map(|(((key, _), time, diff), value)| ((*key, value, time.clone()), *diff)).collect(); + // What the chunks hold, netted at the compared values. + let mut expected = BTreeMap::new(); + for ((key, value, time), diff) in netted(&chunks, &keys.iter().copied().collect()) { + let value = if projected { value[..1].to_vec() } else { value }; + *expected.entry((key, value, time)).or_insert(0) += diff; + } + expected.retain(|_, diff| *diff != 0); + assert_eq!(actual, expected); + } + } + } + /// Corrections net desired against current output by value, whichever ids hold it: two ids of /// one value count together, and a new value gets a new id naming its row. #[test] @@ -926,9 +1200,14 @@ mod tests { let counts: Vec = held.iter().filter(|(key, _)| within(**key)).map(|(_, count)| *count).collect(); assert!(counts.iter().rev().skip(1).sum::() < window_size, "a window stops once it holds the budget"); assert!(from == KeyPosition::End || counts.iter().sum::() >= window_size, "a window holds the budget"); - for (bridge, pool, into) in [(&window.input, &backend.input, &mut inputs), (&window.output, &backend.output, &mut outputs)] { - let values = rows_of(&pool.gather(&bridge.iter().map(|r| r.0.1).collect::>())); - for (((key, _), time, diff), value) in bridge.iter().zip(values) { + // The values the window's ids name, whether rows or the values themselves. + let input_values = backend.input_values(window.input.iter().map(|r| r.0.1)); + let output_ids: Vec = window.output.iter().map(|r| r.0.1).collect(); + let all = 0..output_ids.len(); + let named = if backend.window_ids.output_values { std::slice::from_ref(&all) } else { &[] }; + let output_values = backend.output_values(&output_ids, named); + for (bridge, values, into) in [(&window.input, input_values, &mut inputs), (&window.output, output_values, &mut outputs)] { + for (((key, _), time, diff), value) in bridge.iter().zip(rows_of(&values)) { assert!(within(*key)); assert!(into.insert((*key, value, time.clone()), *diff).is_none()); } @@ -1013,4 +1292,107 @@ mod tests { } } } + + /// Integer values named by themselves reduce as values named by rows do. Over two retires of + /// signed values with ties and cancellations, bare, in a one-field product, or as the field a + /// weighted sum reads of a pair (whose projection is the integer), each reducer gives what it + /// gives under forged identifiers that all collide, where every window names + /// rows: with integer keys, hashed keys, and forged identifiers of which only one collides, + /// so that some windows name values and others rows, at every window size. + #[test] + fn integer_values_reduce_as_rows_do() { + use timely::progress::{Antichain, Timestamp}; + use differential_dataflow::operators::int_proxy::reduce::ProxyReduceTactic; + use differential_dataflow::operators::reduce::ReduceTactic; + use crate::runtime::feed::print; + use crate::parse::Reducer as Sugar; + + /// How keys `(a, 7a)` are identified. + #[derive(Clone, Copy, Debug)] + enum Ident { Int, Hash, OneCollides, AllCollide } + let key_column = |ident: Ident, keys: &[u64]| -> CValue { + let pair = CValue::Prod(vec![ints(keys.to_vec()), ints(keys.iter().map(|k| k * 7).collect())]); + match ident { + Ident::Int => ints(keys.to_vec()), + Ident::Hash => present_key(pair), + Ident::OneCollides => CValue::Prod(vec![ints(keys.iter().map(|&k| if k < 2 { 0 } else { k }).collect()), pair]), + Ident::AllCollide => CValue::Prod(vec![ints(keys.iter().map(|k| k % 2).collect()), pair]), + } + }; + // Signed values at the edges of the sign flip and of byte storage, with ties. + const VALUES: [i64; 8] = [i64::MIN, -300, -1, 0, 1, 255, 256, i64::MAX]; + // Each value with a second field, which a pair's projection drops: values equal in their + // first field may differ in it, and merge only when projected. + let signed = |seed, count| -> Vec<(u64, (i64, i64), Time, Diff)> { + random_rows(seed, count, 6).into_iter().map(|(k, (a, b), t, d)| (k, (VALUES[(a * 3 + b) as usize % 8], a as i64 - 1), t, d)).collect() + }; + let stamp = |outer| crate::server::host_time(outer); + let later = |t: &Time| Time::new(std::iter::once(t.first().copied().unwrap_or(0) + 2).chain(t.iter().skip(1).copied()).collect()); + // The first retire's input, at outer times 0 and 1. The second's, from 2 on: new rows, + // rows of the first negated, and a batch cancelling some new rows exactly. + let early: Vec<_> = signed(0x51ED_270B_2A4F_3C11, 200).into_iter().filter(|r| r.2.first().copied().unwrap_or(0) < 2).collect(); + let mut late: Vec<_> = signed(0x0BAD_5EED, 120).into_iter().map(|(k, v, t, d)| (k, v, later(&t), d)).collect(); + late.extend(early.iter().step_by(7).map(|(k, v, t, d)| (*k, *v, later(t), -d))); + let cancel: Vec<_> = late.iter().step_by(3).map(|(k, v, t, d)| (*k, *v, t.clone(), -d)).collect(); + + /// How the values are held. + #[derive(Clone, Copy, Debug)] + enum Values { Bare, Wrapped, Pair } + for values in [Values::Bare, Values::Wrapped, Values::Pair] { + // Bare and wrapped: each value, and a constant for each, with its diff, so that equal + // outputs consolidate. A pair: the weighted sum of first fields, which reads only them. + let body = match values { + Values::Bare => "(vals map ((v, d) -> (v, d)), vals map ((_, d) -> (7, d))) append", + Values::Wrapped => "(vals map ((v, d) -> (v, d)), vals map ((_, d) -> ((7), d))) append", + Values::Pair => "let n = vals map (((v, _), d) -> (v, d) mul) fold_add in (n, (n, 1)) enlist filter", + }; + let lambda = crate::ir::Lambda::new(format!("let (_, vals) = input in {body}"), &corgi::Scope::default()); + let (vshape, column): (Shape, fn(&[(i64, i64)]) -> CValue) = match values { + Values::Bare => (Shape::Int, |vs| CValue::i64(vs.iter().map(|v| v.0).collect())), + Values::Wrapped => (Shape::Prod(vec![Shape::Int]), |vs| CValue::Prod(vec![CValue::i64(vs.iter().map(|v| v.0).collect())])), + Values::Pair => (Shape::Prod(vec![Shape::Int, Shape::Int]), |vs| CValue::Prod(vec![CValue::i64(vs.iter().map(|v| v.0).collect()), CValue::int(vs.iter().map(|v| v.1))])), + }; + // The sum compares its pairs' projections, single integers, so they name themselves. + if let Values::Pair = values { + let backend = CorgiReduceBackend::