Skip to content

feat: Convergence criterion as a plan sequence (#1834) - #1834

Open
Yuhta wants to merge 1 commit into
facebookincubator:mainfrom
Yuhta:export-D118870175
Open

feat: Convergence criterion as a plan sequence (#1834)#1834
Yuhta wants to merge 1 commit into
facebookincubator:mainfrom
Yuhta:export-D118870175

Conversation

@Yuhta

@Yuhta Yuhta commented Sep 4, 2026

Copy link
Copy Markdown
Contributor

Summary:
X-link: facebookincubator/velox#18854

PageRank's stopping rule is the RMSE of the per-vertex rank change, measured
after the update. That statistic does not exist until the body's last plan
has produced the new ranks -- and that plan's output is the state the
iteration commits, so the reduce that turns per-worker partials into one
verdict has nowhere to go inside the body.

Reducing it needs a shuffle of its own: each worker reduces its shard, the
partials exchange, one final aggregation emits the verdict. A Velox plan
fragment ends at a shuffle boundary, so that is not one plan but a chain of
them -- which a single ConvergenceConfig::plan cannot express.

ConvergenceConfig::plan becomes plans, chained exactly as
FixedPointNode::plans() are: the first starts with a StateSourceNode, every
later one starts with an Exchange, every non-last one ends with a
PartitionedOutput, and the last emits the single BOOLEAN verdict.
converging(plan, n) still takes one plan and delegates to the vector
overload, so existing callers are unchanged.

FixedPointNode validates that the chain really is that: exactly one incoming
shuffle per fragment, so a branching topology such as a distributed join is
rejected rather than silently half-checked on the primary input chain only;
matching schemas across each shuffle boundary; one partition count across every
stage, since taking the maximum does not make incompatible counts safe; and the
last plan's single BOOLEAN column. numWorkers() and requiresSplits()
account for the convergence chain because a convergence sequence that
reduces across workers needs coordinator-assigned peers just as a shuffling
body does: consult the body alone and a non-shuffling body with a shuffling
convergence chain reports one worker and no split requirement, leaving its
exchanges waiting on peers nobody assigned. Serialization round-trips the
chain.

Also adds ConvergenceConfig::whenDeltaEmpty(maxIterations), the semi-naive
termination test -- stop on the first iteration that writes no rows -- which
needs no convergence plan at all, and so no sub-task per iteration to recompute
a row count the framework already holds. Restricted to a non-shuffling fixed
point, because the delta is a worker's local shard.

Worker propagation through a nested fixed point is split into the next diff.

Plan-node half only; the execution that consumes it is at the top of the stack.

Differential Revision: D118870175

@meta-codesync

meta-codesync Bot commented Sep 4, 2026

Copy link
Copy Markdown

@Yuhta has exported this pull request. If you are a Meta employee, you can view the originating Diff in D118870175.

@meta-cla meta-cla Bot added the CLA Signed This label is managed by the Meta Open Source bot. label Sep 4, 2026
Summary:
X-link: facebookincubator/velox#18854


PageRank's stopping rule is the RMSE of the per-vertex rank change, measured
*after* the update.  That statistic does not exist until the body's last plan
has produced the new ranks -- and that plan's output *is* the state the
iteration commits, so the reduce that turns per-worker partials into one
verdict has nowhere to go inside the body.

Reducing it needs a shuffle of its own: each worker reduces its shard, the
partials exchange, one final aggregation emits the verdict.  A Velox plan
fragment ends at a shuffle boundary, so that is not one plan but a chain of
them -- which a single `ConvergenceConfig::plan` cannot express.

`ConvergenceConfig::plan` becomes `plans`, chained exactly as
`FixedPointNode::plans()` are: the first starts with a `StateSourceNode`, every
later one starts with an `Exchange`, every non-last one ends with a
`PartitionedOutput`, and the last emits the single BOOLEAN verdict.
`converging(plan, n)` still takes one plan and delegates to the vector
overload, so existing callers are unchanged.

`FixedPointNode` validates that the chain really is that: exactly one incoming
shuffle per fragment, so a branching topology such as a distributed join is
rejected rather than silently half-checked on the primary input chain only;
matching schemas across each shuffle boundary; one partition count across every
stage, since taking the maximum does not make incompatible counts safe; and the
last plan's single BOOLEAN column.  `numWorkers()` and `requiresSplits()`
account for the convergence chain because a convergence sequence that
reduces across workers needs coordinator-assigned peers just as a shuffling
body does: consult the body alone and a non-shuffling body with a shuffling
convergence chain reports one worker and no split requirement, leaving its
exchanges waiting on peers nobody assigned.  Serialization round-trips the
chain.

Also adds `ConvergenceConfig::whenDeltaEmpty(maxIterations)`, the semi-naive
termination test -- stop on the first iteration that writes no rows -- which
needs no convergence plan at all, and so no sub-task per iteration to recompute
a row count the framework already holds.  Restricted to a non-shuffling fixed
point, because the delta is a worker's local shard.

Worker propagation through a *nested* fixed point is split into the next diff.

Plan-node half only; the execution that consumes it is at the top of the stack.

Differential Revision: D118870175
@meta-codesync meta-codesync Bot changed the title feat: Convergence criterion as a plan sequence feat: Convergence criterion as a plan sequence (#1834) Sep 11, 2026
Yuhta added a commit to Yuhta/velox that referenced this pull request Sep 11, 2026
Summary:

X-link: facebookincubator/axiom#1834

PageRank's stopping rule is the RMSE of the per-vertex rank change, measured
*after* the update.  That statistic does not exist until the body's last plan
has produced the new ranks -- and that plan's output *is* the state the
iteration commits, so the reduce that turns per-worker partials into one
verdict has nowhere to go inside the body.

Reducing it needs a shuffle of its own: each worker reduces its shard, the
partials exchange, one final aggregation emits the verdict.  A Velox plan
fragment ends at a shuffle boundary, so that is not one plan but a chain of
them -- which a single `ConvergenceConfig::plan` cannot express.

`ConvergenceConfig::plan` becomes `plans`, chained exactly as
`FixedPointNode::plans()` are: the first starts with a `StateSourceNode`, every
later one starts with an `Exchange`, every non-last one ends with a
`PartitionedOutput`, and the last emits the single BOOLEAN verdict.
`converging(plan, n)` still takes one plan and delegates to the vector
overload, so existing callers are unchanged.

`FixedPointNode` validates that the chain really is that: exactly one incoming
shuffle per fragment, so a branching topology such as a distributed join is
rejected rather than silently half-checked on the primary input chain only;
matching schemas across each shuffle boundary; one partition count across every
stage, since taking the maximum does not make incompatible counts safe; and the
last plan's single BOOLEAN column.  `numWorkers()` and `requiresSplits()`
account for the convergence chain because a convergence sequence that
reduces across workers needs coordinator-assigned peers just as a shuffling
body does: consult the body alone and a non-shuffling body with a shuffling
convergence chain reports one worker and no split requirement, leaving its
exchanges waiting on peers nobody assigned.  Serialization round-trips the
chain.

Also adds `ConvergenceConfig::whenDeltaEmpty(maxIterations)`, the semi-naive
termination test -- stop on the first iteration that writes no rows -- which
needs no convergence plan at all, and so no sub-task per iteration to recompute
a row count the framework already holds.  Restricted to a non-shuffling fixed
point, because the delta is a worker's local shard.

Worker propagation through a *nested* fixed point is split into the next diff.

Plan-node half only; the execution that consumes it is at the top of the stack.

Differential Revision: D118870175
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

CLA Signed This label is managed by the Meta Open Source bot. meta-exported

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant