Skip to content

DDIR reduce: integer values are their own ids - #1068

Closed
frankmcsherry wants to merge 1 commit into
TimelyDataflow:master-nextfrom
frankmcsherry:ddir-int-value-ids
Closed

frankmcsherry wants to merge 1 commit into
TimelyDataflow:master-nextfrom
frankmcsherry:ddir-int-value-ids

Conversation

@frankmcsherry

Copy link
Copy Markdown
Member

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 inside one-field products, now names itself: its bits with the sign bit flipped, so ids ascend in value order and equal values share an id without a sort. One commit, in interactive/src/runtime/reduce.rs (+429 −47, about half tests).

What changes on the fast path.

  • Numbering: ids are read from the borrowed i64 lanes. There is no gather, no sort_blocks, and no pool entry per value. A key held by several chunks gets a stable sort of its ids.
  • The reduction's input: the value column is rebuilt from the ids.
  • Corrections: a key's desired ids minus its output ids, consolidated. Nothing is matched by value comparison, and nothing is pushed to the output pool.
  • Emission: the ids are decoded back into the value column.

When it applies.

  • Input: by the values present compares, which since DDIR: project values in the first phase of join, half join and reduce #1058 is their projection where a reduction is proven to merge equal projections. A pair read at one integer field projects to (int) and names itself. count projects to a unit and keeps row ids. Projected fields held as bytes are gathered before their integers are read.
  • Output: by the reduction's output value shape.
  • Per window: only where each identifier holds one real key. That is always for integer keys, and for hashed keys unless the window holds a 64-bit hash collision; such a window keeps pool ids. A batch built from both kinds of window gathers from the pool and the decoded values together.

Measured with a probe of the reduce backend, each scenario in its own process, base and branch alternating, median of five launches:

Data Presentation Churn retire Initial retire
SCC-like, int key, min, 1% churn −7% −2% −4%
SCC-like, hashed key, min, 1% churn −18% −17% −4%
SCC-like, int key, min, 10% churn −14% −9% −4%
Index-like, int key, distinct −26% −63% −45%
Index-like, int key, (int, int) value summed at its first field −19% −24% −4%

The count scenarios present one row per key under #1058 (1,000 rows from 126k history rows), and the base's timings for them swing between 1.65 and 3.0 ms from launch to launch, so they show no real change either way.

One shared change. present now builds each window's labels and rows at their final size, for both kinds of id. Grown by doubling, they cost a presentation up to half again its time depending on the allocator's state: the projected pair sum ran at 3.1 or 2.1 ms by the launch, and at 2.0 ms with the reservation in 8 of 8 launches.

The join is unchanged: it already uses integer values as their own tokens, and it numbers no values with a segmented sort.

Checks: cargo test -p differential-dataflow -p differential-dogs3 -p interactive -p ddir-server (490 passed) and the 1,000-case release difftest. The new end-to-end test runs count, distinct, min, collect and a weighted sum over bare, one-field-product and pair values, with ties, retractions and i64::MIN/i64::MAX, and requires each to equal the run where every identifier collides, which forces pool ids everywhere.

Implemented and measured by a Claude subagent.

🤖 Generated with Claude Code

https://claude.ai/code/session_019XsakwFGwtESwZ1gSYwG4o

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) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019XsakwFGwtESwZ1gSYwG4o
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant