diff --git a/velox/core/FixedPointPlanNodes.cpp b/velox/core/FixedPointPlanNodes.cpp index 0f2bc1d573d..33a16cce1ba 100644 --- a/velox/core/FixedPointPlanNodes.cpp +++ b/velox/core/FixedPointPlanNodes.cpp @@ -30,6 +30,24 @@ const PlanNode* primaryLeaf(const PlanNode* node) { return node; } +// Whether 'node' is, or reaches, a nested FixedPointNode. Recursion stops at +// one because FixedPointNode::sources() is empty: its plans are sub-tasks, not +// pipeline inputs. +bool containsNestedFixedPoint(const PlanNodePtr& node) { + if (node == nullptr) { + return false; + } + if (dynamic_cast(node.get()) != nullptr) { + return true; + } + for (const auto& source : node->sources()) { + if (containsNestedFixedPoint(source)) { + return true; + } + } + return false; +} + // Returns true if 'node' or any of its sources requires splits (e.g. a // TableScan or an Exchange) -- i.e. the coordinator must assign it source // splits. @@ -132,16 +150,50 @@ FixedPointNode::FixedPointNode( convergenceConfig_.maxIterations, 0, "FixedPointNode: maxIterations must be positive"); + // Contradictions in the config itself are reported before anything validates + // the structure they describe, so the message names the contradiction rather + // than a consequence of it. + VELOX_USER_CHECK( + !convergenceConfig_.stopWhenDeltaEmpty || + convergenceConfig_.plans.empty(), + "FixedPointNode: stopWhenDeltaEmpty and a convergence sequence are " + "mutually exclusive; the delta row count is already the verdict"); validatePlans(); + validateWorkerCounts(); resolveAndValidateStateReferences(); // errorWhenMaxIterationReached fails the loop when maxIterations is reached - // without converging; that is only meaningful with a convergence plan (a - // null plan never converges, so it would always fail). + // without converging; that is only meaningful with a convergence criterion + // (with none the loop never converges, so it would always fail). VELOX_USER_CHECK( !convergenceConfig_.errorWhenMaxIterationReached || - convergenceConfig_.plan != nullptr, + !convergenceConfig_.plans.empty() || + convergenceConfig_.stopWhenDeltaEmpty, "FixedPointNode: errorWhenMaxIterationReached requires a convergence " - "plan; set it false for a fixed-count loop with no convergence plan"); + "criterion; set it false for a fixed-count loop with no convergence " + "plan"); + // The delta is the local shard's row count and the framework performs no + // cross-worker reduction, so a peer's frontier may still be non-empty when + // this one empties -- stopping then strands the peers reading this worker. + if (convergenceConfig_.stopWhenDeltaEmpty) { + VELOX_USER_CHECK_EQ( + numWorkers(), + 1, + "FixedPointNode: stopWhenDeltaEmpty requires a non-shuffling fixed " + "point, because the delta is local to a worker"); + // numWorkers() does not yet see through a nested FixedPointNode, so a + // nested loop that shuffles would leave this reporting one worker and the + // check above satisfied. Reject any nested loop rather than accept the + // case the check exists to reject; the follow-up that propagates the + // nested width replaces this with the width comparison. + for (const auto& chain : {plans_, convergenceConfig_.plans}) { + for (const auto& plan : chain) { + VELOX_USER_CHECK( + !containsNestedFixedPoint(plan), + "FixedPointNode: stopWhenDeltaEmpty does not support a nested " + "fixed point, whose width this node cannot yet see"); + } + } + } } void FixedPointNode::addDetails(std::stringstream& stream) const { @@ -151,20 +203,57 @@ void FixedPointNode::addDetails(std::stringstream& stream) const { << convergenceConfig_.toString(); } -bool FixedPointNode::requiresSplits() const { - if (!plans_.empty() && +namespace { + +// Whether a plan chain makes the fixed point need coordinator-assigned peer +// splits: its first plan shuffles out, or a later plan shuffles in (an +// Exchange) and so consumes splits. Without this a shuffling chain would +// report requiresSplits()==false and deadlock waiting for peers that were never +// named. +bool chainRequiresSplits(const std::vector& plans) { + if (!plans.empty() && std::dynamic_pointer_cast( - plans_.front()) != nullptr) { + plans.front()) != nullptr) { return true; } - // A non-first body plan that shuffles in (an Exchange) consumes splits the - // coordinator must assign; without this a multi-plan shuffling fixed point - // would report requiresSplits()==false and deadlock waiting for peers. - for (size_t i = 1; i < plans_.size(); ++i) { - if (containsSplitSource(plans_[i])) { + for (size_t i = 1; i < plans.size(); ++i) { + if (containsSplitSource(plans[i])) { return true; } } + return false; +} + +// The number of workers a chain shuffles across -- the partition count its +// first plan's PartitionedOutput writes to -- or 1 when the chain does not +// shuffle. +int32_t chainWorkers(const std::vector& plans) { + if (plans.empty()) { + return 1; + } + if (auto partitioned = + std::dynamic_pointer_cast( + plans.front())) { + return partitioned->numPartitions(); + } + return 1; +} + +} // namespace + +int32_t FixedPointNode::numWorkers() const { + // validateWorkerCounts() rejected chains that disagree, so these are either + // equal or one of them is 1 -- the max is the single agreed width either way. + return std::max(chainWorkers(plans_), chainWorkers(convergenceConfig_.plans)); +} + +bool FixedPointNode::requiresSplits() const { + // Either chain can be the shuffling one -- a convergence sequence that + // reduces across workers needs peers just as a shuffling body does. + if (chainRequiresSplits(plans_) || + chainRequiresSplits(convergenceConfig_.plans)) { + return true; + } for (const auto& declaration : stateDeclarations_) { if (containsSplitSource(declaration->initialPlan())) { return true; @@ -173,53 +262,161 @@ bool FixedPointNode::requiresSplits() const { return false; } -void FixedPointNode::validatePlans() const { - VELOX_USER_CHECK( - !plans_.empty(), "FixedPointNode requires at least one plan"); - const auto numPlans = plans_.size(); +namespace { + +// Validates the sub-plan chaining convention, which the body and convergence +// sequences share: the first plan reads state, every later one receives the +// previous one's shuffle through an Exchange, and only the last produces rows +// instead of shuffling through a PartitionedOutput. 'what' names a plan of the +// sequence and 'lastPlanProduces' what its last plan emits, both for the error +// messages. +// Counts the ExchangeNodes anywhere in 'node', not just on its primary input +// chain. A fragment reading a second shuffle is a branching topology (a +// distributed join, say) rather than one link of a linear chain. +int32_t countExchanges(const PlanNode* node) { + if (node == nullptr) { + return 0; + } + int32_t numExchanges = + dynamic_cast(node) != nullptr ? 1 : 0; + for (const auto& source : node->sources()) { + numExchanges += countExchanges(source.get()); + } + return numExchanges; +} + +void validatePlanChain( + const std::vector& plans, + std::string_view what, + std::string_view lastPlanProduces) { + const auto numPlans = plans.size(); for (size_t i = 0; i < numPlans; ++i) { VELOX_USER_CHECK_NOT_NULL( - plans_[i], "FixedPointNode: plan {} must not be null", i); - const auto* root = plans_[i].get(); + plans[i], "FixedPointNode: {} {} must not be null", what, i); + const auto* root = plans[i].get(); const auto* leaf = primaryLeaf(root); const std::string leafName = leaf != nullptr ? std::string(leaf->name()) : "nothing"; - // The first plan reads from state; every later plan receives the previous - // plan's shuffle through an Exchange. + // The chain is one logical plan cut at its shuffle boundaries, so a + // fragment reads at most the one shuffle that links it to its predecessor. + // Counting every Exchange, not just the one on the primary input chain, + // is what rejects a branching fragment. + const auto numExchanges = countExchanges(root); if (i == 0) { VELOX_USER_CHECK( dynamic_cast(leaf) != nullptr, - "FixedPointNode: the first plan must start with a StateSourceNode, " + "FixedPointNode: the first {} must start with a StateSourceNode, " "but it starts with {}", + what, leafName); + VELOX_USER_CHECK_EQ( + numExchanges, + 0, + "FixedPointNode: the first {} must not read a shuffle", + what); } else { VELOX_USER_CHECK( dynamic_cast(leaf) != nullptr, - "FixedPointNode: every non-first plan must start with an Exchange, " - "but plan {} starts with {}", + "FixedPointNode: every non-first {} must start with an Exchange, " + "but {} {} starts with {}", + what, + what, i, leafName); + VELOX_USER_CHECK_EQ( + numExchanges, + 1, + "FixedPointNode: every non-first {} must read exactly one shuffle, " + "the link to its predecessor; a branching topology such as a " + "distributed join is not supported. Plan: {}", + what, + i); } - // The last plan produces the rows the framework writes back to the output - // state entry, so it must not end with a PartitionedOutput; every earlier - // plan shuffles to the next through one. + const std::string rootName = + root != nullptr ? std::string(root->name()) : "nothing"; if (i + 1 == numPlans) { VELOX_USER_CHECK( dynamic_cast(root) == nullptr, - "FixedPointNode: the last plan must produce the rows written back to " - "the output state entry, not shuffle through a PartitionedOutput, but " - "plan {} ends with {}", + "FixedPointNode: the last {} must produce {}, not shuffle through a " + "PartitionedOutput, but {} {} ends with {}", + what, + lastPlanProduces, + what, i, - root != nullptr ? std::string(root->name()) : "nothing"); + rootName); } else { VELOX_USER_CHECK( dynamic_cast(root) != nullptr, - "FixedPointNode: every non-last plan must end with a " - "PartitionedOutput, but plan {} ends with {}", + "FixedPointNode: every non-last {} must end with a " + "PartitionedOutput, but {} {} ends with {}", + what, + what, i, - root != nullptr ? std::string(root->name()) : "nothing"); + rootName); } } + + // Adjacent fragments are the two halves of one shuffle: what the + // PartitionedOutput writes is what the next Exchange reads, so their schemas + // must match, and every stage must shuffle across the same width or the + // exchanges are built for peers that were never named. + const auto workers = chainWorkers(plans); + for (size_t i = 0; i + 1 < numPlans; ++i) { + const auto partitioned = + std::dynamic_pointer_cast(plans[i]); + VELOX_USER_CHECK_EQ( + partitioned->numPartitions(), + workers, + "FixedPointNode: every shuffling {} must partition across the same " + "number of workers. Plan: {}", + what, + i); + const auto* exchange = dynamic_cast( + primaryLeaf(plans[i + 1].get())); + VELOX_USER_CHECK( + exchange->outputType()->equivalent(*plans[i]->outputType()), + "FixedPointNode: the schema a {} shuffles out must match what the next " + "one reads back. Plan {} writes {}, plan {} reads {}", + what, + i, + plans[i]->outputType()->toString(), + i + 1, + exchange->outputType()->toString()); + } +} + +} // namespace + +void FixedPointNode::validatePlans() const { + VELOX_USER_CHECK( + !plans_.empty(), "FixedPointNode requires at least one plan"); + validatePlanChain( + plans_, "plan", "the rows written back to the output state entry"); + // The convergence sequence chains like the body -- reading state, shuffling + // between plans -- so the same check applies. Structure is checked before + // validateWorkerCounts() compares widths, so a malformed chain reports the + // structural error rather than a worker-count mismatch derived from it. + if (!convergenceConfig_.plans.empty()) { + validatePlanChain( + convergenceConfig_.plans, + "convergence plan", + "the BOOLEAN convergence verdict"); + } +} + +void FixedPointNode::validateWorkerCounts() const { + const auto bodyWorkers = chainWorkers(plans_); + const auto convergenceWorkers = chainWorkers(convergenceConfig_.plans); + // A chain that does not shuffle expresses no opinion. Two that do must + // agree: the body's exchanges are wired for its own peer count, so a wider + // convergence chain would wait on peers the coordinator never assigned. + VELOX_USER_CHECK( + bodyWorkers == 1 || convergenceWorkers == 1 || + bodyWorkers == convergenceWorkers, + "FixedPointNode: the body and convergence chains must shuffle across the " + "same number of workers. Body: {}, convergence: {}", + bodyWorkers, + convergenceWorkers); } void FixedPointNode::resolveAndValidateStateReferences() const { @@ -237,9 +434,10 @@ void FixedPointNode::resolveAndValidateStateReferences() const { // Resolve and check every StateSource / StateHashJoin in every body plan and // in the convergence plan (which also reads state via a StateSource). std::vector referencingPlans = plans_; - if (convergenceConfig_.plan != nullptr) { - referencingPlans.push_back(convergenceConfig_.plan); - } + referencingPlans.insert( + referencingPlans.end(), + convergenceConfig_.plans.begin(), + convergenceConfig_.plans.end()); for (const auto& plan : referencingPlans) { std::vector sources; collectNodes(plan, sources); @@ -365,16 +563,10 @@ void FixedPointNode::resolveAndValidateStateReferences() const { "schema for entry: {}", outputStateEntry_); - // The convergence plan (when present) starts with a StateSourceNode and emits - // exactly one BOOLEAN column. - if (convergenceConfig_.plan != nullptr) { - const auto* leaf = primaryLeaf(convergenceConfig_.plan.get()); - VELOX_USER_CHECK( - dynamic_cast(leaf) != nullptr, - "FixedPointNode: the convergence plan must start with a StateSourceNode," - " but it starts with {}", - leaf != nullptr ? std::string(leaf->name()) : "nothing"); - const auto& convergenceType = convergenceConfig_.plan->outputType(); + // The convergence sequence's last plan emits exactly one BOOLEAN column, the + // verdict. Its chain structure is checked in validatePlans(). + if (!convergenceConfig_.plans.empty()) { + const auto& convergenceType = convergenceConfig_.plans.back()->outputType(); VELOX_USER_CHECK_EQ( convergenceType->size(), 1, @@ -517,38 +709,58 @@ HashTableState& HashTableState::initial(PlanNodePtr initialPlan) { // static ConvergenceConfig ConvergenceConfig::withMaxIterations(int32_t maxIterations) { return ConvergenceConfig{ - .plan = nullptr, + .plans = {}, .maxIterations = maxIterations, .errorWhenMaxIterationReached = false}; } +// static +ConvergenceConfig ConvergenceConfig::whenDeltaEmpty(int32_t maxIterations) { + return ConvergenceConfig{ + .plans = {}, + .maxIterations = maxIterations, + .errorWhenMaxIterationReached = true, + .stopWhenDeltaEmpty = true}; +} + // static ConvergenceConfig ConvergenceConfig::converging( PlanNodePtr plan, int32_t maxIterations) { + return converging(std::vector{std::move(plan)}, maxIterations); +} + +// static +ConvergenceConfig ConvergenceConfig::converging( + std::vector plans, + int32_t maxIterations) { return ConvergenceConfig{ - .plan = std::move(plan), + .plans = std::move(plans), .maxIterations = maxIterations, .errorWhenMaxIterationReached = true}; } folly::dynamic ConvergenceConfig::serialize() const { folly::dynamic obj = folly::dynamic::object; - if (plan != nullptr) { - obj["plan"] = plan->serialize(); + folly::dynamic serializedPlans = folly::dynamic::array; + for (const auto& plan : plans) { + serializedPlans.push_back(plan->serialize()); } + obj["plans"] = std::move(serializedPlans); obj["maxIterations"] = maxIterations; obj["errorWhenMaxIterationReached"] = errorWhenMaxIterationReached; + obj["stopWhenDeltaEmpty"] = stopWhenDeltaEmpty; return obj; } std::string ConvergenceConfig::toString() const { return fmt::format( "maxIterations: {}, errorWhenMaxIterationReached: {}, " - "convergencePlan: {}", + "stopWhenDeltaEmpty: {}, convergencePlans: {}", maxIterations, errorWhenMaxIterationReached ? "true" : "false", - plan != nullptr ? "present" : "none"); + stopWhenDeltaEmpty ? "true" : "false", + plans.size()); } // static @@ -556,12 +768,13 @@ ConvergenceConfig ConvergenceConfig::deserialize( const folly::dynamic& obj, void* context) { ConvergenceConfig config; - if (obj.count("plan") != 0u) { - config.plan = ISerializable::deserialize(obj["plan"], context); + for (const auto& plan : obj["plans"]) { + config.plans.push_back(ISerializable::deserialize(plan, context)); } config.maxIterations = static_cast(obj["maxIterations"].asInt()); config.errorWhenMaxIterationReached = obj["errorWhenMaxIterationReached"].asBool(); + config.stopWhenDeltaEmpty = obj["stopWhenDeltaEmpty"].asBool(); return config; } diff --git a/velox/core/FixedPointPlanNodes.h b/velox/core/FixedPointPlanNodes.h index 37c02918f50..1a834c64a32 100644 --- a/velox/core/FixedPointPlanNodes.h +++ b/velox/core/FixedPointPlanNodes.h @@ -227,48 +227,84 @@ class HashTableState { /// terminal state). maxIterations (a ConvergenceConfig field below) is always /// active as a safety bound. /// -/// Multi-worker contract: each worker evaluates `plan` over its own local +/// Multi-worker contract: each worker evaluates the sequence over its own local /// shard, so for a shuffling fixed point (N > 1) the verdict must be globally /// consistent -- every worker must reach the same value on the same iteration, /// or lockstep breaks and the shuffle deadlocks. Making it consistent is the -/// plan's responsibility: synchronize the convergence-deciding state across -/// workers through the body's shuffle (e.g. replicate it), so each worker's -/// local read agrees. The framework deliberately performs no cross-worker -/// reduction (that would add an all-reduce shuffle). A null plan means "never -/// converge" (the loop is bounded only by maxIterations) and is always safe. +/// plan's responsibility, and there are two ways to do it. When the body's +/// shuffle already replicates the convergence-deciding state, a single-plan +/// sequence reads it locally and every worker agrees. When the statistic is +/// only known after the body's last exchange -- PageRank's post-update RMSE, +/// say, whose squared error does not exist until the new ranks are computed, +/// and whose reduce therefore has nowhere to go in the body (the last body plan +/// produces the state itself) -- the sequence shuffles: earlier plans reduce +/// partials, the last emits the verdict. The framework performs no +/// cross-worker reduction of its own. An empty sequence means "never converge" +/// (the loop is bounded only by maxIterations) and is always safe. struct ConvergenceConfig { - /// Plan producing exactly one BOOLEAN output column. Starts with a - /// StateSourceNode. The framework reads that single column after each - /// iteration; a true value stops the loop. Null means never converge (loop - /// bounded by maxIterations). - PlanNodePtr plan{nullptr}; + /// Plans run in order after each iteration, chained by shuffle exactly as + /// FixedPointNode::plans() are: the first starts with a StateSourceNode + /// reading the state the iteration just committed, every later one starts + /// with an Exchange, every non-last one ends with a PartitionedOutput. The + /// last plan produces exactly one BOOLEAN output column, which the framework + /// reads; true stops the loop. Convergence plans never write state. Empty + /// means never converge (loop bounded by maxIterations). + std::vector plans{}; /// Maximum iterations the loop runs -- always active as a safety bound, and - /// the sole bound when `plan` is null (a fixed-count loop). + /// the sole bound when `plans` is empty (a fixed-count loop). int32_t maxIterations{0}; /// Decides what happens when `maxIterations` is reached before the - /// convergence `plan` signals convergence. When true (the default), the + /// convergence sequence signals convergence. When true (the default), the /// fixed point fails -- a guard against silently returning a non-converged /// result. When false, the loop instead stops at the cap and returns the /// current approximate result without failing: best-effort convergence (run /// toward convergence, but accept the partial result if the cap is hit - /// first), as in an early-stopped KMeans. Requires a convergence `plan` - /// (FixedPointNode validates this) -- a null plan never converges, so set - /// this false for a fixed-count loop with no convergence plan, where the cap - /// is the intended stopping point rather than a best-effort cutoff. + /// first), as in an early-stopped KMeans or PageRank. Requires a non-empty + /// `plans` (FixedPointNode validates this) -- an empty sequence never + /// converges, so set this false for a fixed-count loop, where the cap is the + /// intended stopping point rather than a best-effort cutoff. bool errorWhenMaxIterationReached{true}; - /// Builds a fixed-count loop config: no convergence plan, bounded only by + /// Stops the loop on the first iteration that writes no rows into the output + /// state entry -- the empty-frontier terminal state -- without running a + /// convergence sequence. Mutually exclusive with `plans`: the verdict is a + /// row count the framework already holds, so expressing it as a plan would + /// spend a sub-task per iteration recomputing what the body just produced. + /// Restricted to a non-shuffling fixed point, because the delta is this + /// worker's local shard and the framework performs no cross-worker + /// reduction: with peers, one worker's frontier can empty while others still + /// read its output. + bool stopWhenDeltaEmpty{false}; + + /// Builds a fixed-count loop config: no convergence plans, bounded only by /// `maxIterations`, so it never errors on reaching the bound. Use for loops /// with a known iteration count (e.g. Fibonacci, a bounded expansion). static ConvergenceConfig withMaxIterations(int32_t maxIterations); - /// Builds a convergence-plan loop config: runs `plan` after each iteration - /// and stops when its BOOLEAN output is true, failing if `maxIterations` is - /// reached first. + /// Builds a delta-empty loop config: stops on the first iteration that + /// produces no rows, failing if `maxIterations` is reached first. Runs no + /// convergence plan. Use for semi-naive recursion -- recursive CTEs, + /// variable-length paths -- where an empty frontier is the termination + /// condition. + static ConvergenceConfig whenDeltaEmpty(int32_t maxIterations); + + /// Builds a convergence loop config: runs `plan` after each iteration and + /// stops when its BOOLEAN output is true, failing if `maxIterations` is + /// reached first. For a convergence criterion that needs its own shuffle, + /// pass the whole chain to the vector overload. static ConvergenceConfig converging(PlanNodePtr plan, int32_t maxIterations); + /// Builds a convergence loop config whose criterion needs its own shuffle: + /// `plans` is one logical plan cut at its shuffle boundaries, chained exactly + /// as FixedPointNode::plans() are, and its last plan emits the single BOOLEAN + /// verdict. Use when the statistic only exists after the body's last + /// exchange, so its reduce cannot travel on the body's shuffle. + static ConvergenceConfig converging( + std::vector plans, + int32_t maxIterations); + folly::dynamic serialize() const; static ConvergenceConfig deserialize( @@ -277,7 +313,8 @@ struct ConvergenceConfig { /// Renders the convergence properties for plan printing /// (FixedPointNode::addDetails): max iteration bound, whether reaching it is - /// an error, and whether a convergence plan is present. + /// an error, whether the delta emptying stops the loop, and how many + /// convergence plans there are. std::string toString() const; }; @@ -346,6 +383,12 @@ class FixedPointNode : public PlanNode { /// fully local fixed point needs no splits. bool requiresSplits() const override; + /// Number of workers this fixed point runs as: one per output partition of + /// whichever chain shuffles -- the body or the convergence sequence. 1 when + /// neither does. A nested FixedPointNode does not widen the enclosing loop + /// yet; that propagation is a follow-up. + int32_t numWorkers() const; + folly::dynamic serialize() const override; static PlanNodePtr create(const folly::dynamic& obj, void* context); @@ -363,6 +406,10 @@ class FixedPointNode : public PlanNode { // sees the complete node types it inspects. void validatePlans() const; + // Rejects a body chain and a convergence chain that shuffle across different + // worker counts; numWorkers() relies on this to take the wider of the two. + void validateWorkerCounts() const; + // Resolves every StateSource / StateHashJoin reference (in the body plans and // the convergence plan) to a declared state entry of the matching kind, and // checks that referenced schemas, probe-key arity, initial-plan and last-plan @@ -385,6 +432,8 @@ class FixedPointNode : public PlanNode { RowTypePtr outputType_; }; +using FixedPointNodePtr = std::shared_ptr; + /// Reads a Vector persistent state entry as row batches into the pipeline. /// /// Always a leaf node (no sources). Used as the first operator in a plan @@ -450,6 +499,8 @@ class StateSourceNode : public PlanNode { bool delta_; }; +using StateSourceNodePtr = std::shared_ptr; + /// Inner-joins the input (probe) against a HashTable persistent state entry /// that is built once and reused across iterations (hash-table reuse). Output /// rows are the probe input columns followed by the hash table's dependent @@ -509,4 +560,6 @@ class StateHashJoinNode : public PlanNode { std::vector sources_; }; +using StateHashJoinNodePtr = std::shared_ptr; + } // namespace facebook::velox::core diff --git a/velox/core/tests/PlanNodeTest.cpp b/velox/core/tests/PlanNodeTest.cpp index da4f251f636..b666c91289e 100644 --- a/velox/core/tests/PlanNodeTest.cpp +++ b/velox/core/tests/PlanNodeTest.cpp @@ -594,6 +594,211 @@ TEST_F(PlanNodeTest, aggregationNodeNoGroupsSpanBatches) { } } +// A convergence or body chain is one logical plan cut at its shuffle +// boundaries. These are the ways a caller can hand over something that is not +// that, each of which would otherwise wire exchanges to peers that do not +// match. +TEST_F(PlanNodeTest, fixedPointChainValidation) { + auto schema = ROW("x", BIGINT()); + auto otherSchema = ROW({"x", "y"}, BIGINT()); + auto declaration = [&] { + return std::make_shared( + "n", schema, /*initialPlan=*/nullptr, /*append=*/true); + }; + auto stateSource = [&] { + return std::make_shared("b", "n", schema, /*delta=*/true); + }; + auto exchange = [&](const std::string& id, const RowTypePtr& type) { + return std::make_shared(id, type, "Presto"); + }; + auto shuffleOut = [&](const std::string& id, + const PlanNodePtr& source, + int32_t numPartitions, + const RowTypePtr& type) { + return std::make_shared( + id, + PartitionedOutputNode::Kind::kPartitioned, + std::vector{ + std::make_shared(BIGINT(), "x")}, + numPartitions, + /*replicateNullsAndAny=*/false, + std::make_shared(), + type, + "Presto", + std::string(TransportKind::kInMemory), + source); + }; + auto fixedPoint = [&](std::vector plans, + ConvergenceConfig convergence) { + return std::make_shared( + "fp", + std::vector{declaration()}, + std::move(plans), + std::move(convergence), + "n"); + }; + auto noConvergence = [] { + return ConvergenceConfig{ + .maxIterations = 5, .errorWhenMaxIterationReached = false}; + }; + + // A two-plan chain shuffling across two workers is the valid baseline each + // case below mutates, and it is what numWorkers()/requiresSplits() report. + { + auto node = fixedPoint( + {shuffleOut("p", stateSource(), 2, schema), exchange("e", schema)}, + noConvergence()); + EXPECT_EQ(node->numWorkers(), 2); + EXPECT_TRUE(node->requiresSplits()); + } + + // A non-shuffling body is one worker and needs no peer splits. + { + auto node = fixedPoint({stateSource()}, noConvergence()); + EXPECT_EQ(node->numWorkers(), 1); + EXPECT_FALSE(node->requiresSplits()); + } + + // A fragment reading a second shuffle is a branching topology, not one link + // of a linear chain. primaryLeaf() alone would not see the second branch. + { + auto branching = std::make_shared( + "lp", + LocalPartitionNode::Type::kGather, + /*scaleWriter=*/false, + std::make_shared(), + std::vector{ + exchange("e0", schema), exchange("e1", schema)}); + VELOX_ASSERT_USER_THROW( + fixedPoint( + {shuffleOut("p", stateSource(), 2, schema), branching}, + noConvergence()), + "must read exactly one shuffle"); + } + + // What one fragment shuffles out is what the next reads back, so the schemas + // must match. + VELOX_ASSERT_USER_THROW( + fixedPoint( + {shuffleOut("p", stateSource(), 2, schema), + exchange("e", otherSchema)}, + noConvergence()), + "must match what the next one reads back"); + + // Every shuffling stage of a chain crosses the same number of workers. + VELOX_ASSERT_USER_THROW( + fixedPoint( + {shuffleOut("p0", stateSource(), 2, schema), + shuffleOut("p1", exchange("e0", schema), 3, schema), + exchange("e1", schema)}, + noConvergence()), + "must partition across the same number of workers"); + + // A convergence chain wider than the body would wait on peers the + // coordinator never assigned it. + { + auto convergenceSchema = ROW("c", BOOLEAN()); + ConvergenceConfig convergence{ + .plans = + {shuffleOut( + "cp", + std::make_shared( + "cs", "n", schema, /*delta=*/false), + 3, + schema), + exchange("ce", schema)}, + .maxIterations = 5}; + VELOX_ASSERT_USER_THROW( + fixedPoint( + {shuffleOut("p", stateSource(), 2, schema), exchange("e", schema)}, + std::move(convergence)), + "must shuffle across the same number of workers"); + } +} + +// stopWhenDeltaEmpty reads a row count local to one worker, so it is only a +// sound verdict when nothing in the loop shuffles -- and it cannot be combined +// with a convergence sequence, which would be a second, disagreeing verdict. +TEST_F(PlanNodeTest, fixedPointDeltaEmptyValidation) { + auto schema = ROW("x", BIGINT()); + auto declaration = [&] { + return std::make_shared( + "n", schema, /*initialPlan=*/nullptr, /*append=*/true); + }; + auto stateSource = [&](const std::string& id) { + return std::make_shared(id, "n", schema, /*delta=*/true); + }; + auto shuffleOut = [&](const std::string& id, const PlanNodePtr& source) { + return std::make_shared( + id, + PartitionedOutputNode::Kind::kPartitioned, + std::vector{ + std::make_shared(BIGINT(), "x")}, + /*numPartitions=*/2, + /*replicateNullsAndAny=*/false, + std::make_shared(), + schema, + "Presto", + std::string(TransportKind::kInMemory), + source); + }; + auto fixedPoint = [&](std::vector plans, + ConvergenceConfig convergence) { + return std::make_shared( + "fp", + std::vector{declaration()}, + std::move(plans), + std::move(convergence), + "n"); + }; + + // A purely local loop is the case it is for. + EXPECT_NO_THROW( + fixedPoint({stateSource("b")}, ConvergenceConfig::whenDeltaEmpty(10))); + + // The row count is already the verdict; a convergence sequence would be a + // second one. + auto withPlans = ConvergenceConfig::whenDeltaEmpty(10); + withPlans.plans = {stateSource("c")}; + VELOX_ASSERT_USER_THROW( + fixedPoint({stateSource("b")}, std::move(withPlans)), + "mutually exclusive"); + + // A shuffling body means peers, and one worker's frontier can empty while + // theirs has not. + VELOX_ASSERT_USER_THROW( + fixedPoint( + {shuffleOut("p", stateSource("b")), + std::make_shared("e", schema, "Presto")}, + ConvergenceConfig::whenDeltaEmpty(10)), + "non-shuffling fixed point"); + + // numWorkers() cannot yet see a nested loop's width, so any nested loop is + // rejected rather than silently treated as single-worker. + auto nested = std::make_shared( + "inner", + std::vector{std::make_shared( + "m", schema, /*initialPlan=*/nullptr, /*append=*/true)}, + std::vector{ + shuffleOut( + "ip", + std::make_shared( + "is", "m", schema, /*delta=*/true)), + std::make_shared("ie", schema, "Presto")}, + ConvergenceConfig{ + .maxIterations = 5, .errorWhenMaxIterationReached = false}, + "m"); + auto beside = std::make_shared( + "g", + LocalPartitionNode::Type::kGather, + /*scaleWriter=*/false, + std::make_shared(), + std::vector{stateSource("b"), nested}); + VELOX_ASSERT_USER_THROW( + fixedPoint({beside}, ConvergenceConfig::whenDeltaEmpty(10)), + "does not support a nested fixed point"); +} + // The FixedPointNode constructor and the state declarations validate their // inputs up front, so a malformed plan fails at construction rather than at // execution. @@ -682,7 +887,7 @@ TEST_F(PlanNodeTest, fixedPointValidation) { ConvergenceConfig{ .maxIterations = 5, .errorWhenMaxIterationReached = true}, "n"), - "errorWhenMaxIterationReached requires a convergence plan"); + "errorWhenMaxIterationReached requires a convergence criterion"); // A convergence plan must emit exactly one BOOLEAN column. auto nonBoolConvergence = @@ -692,7 +897,7 @@ TEST_F(PlanNodeTest, fixedPointValidation) { "fp", std::vector{vectorN()}, std::vector{body}, - ConvergenceConfig{.plan = nonBoolConvergence, .maxIterations = 5}, + ConvergenceConfig{.plans = {nonBoolConvergence}, .maxIterations = 5}, "n"), "convergence plan output column must be BOOLEAN"); @@ -706,7 +911,7 @@ TEST_F(PlanNodeTest, fixedPointValidation) { vectorN(), std::make_shared("flags", twoColSchema)}, std::vector{body}, - ConvergenceConfig{.plan = twoColConvergence, .maxIterations = 5}, + ConvergenceConfig{.plans = {twoColConvergence}, .maxIterations = 5}, "n"), "exactly one output column"); diff --git a/velox/exec/tests/PlanNodeSerdeTest.cpp b/velox/exec/tests/PlanNodeSerdeTest.cpp index 88e695b324e..b1a600ef64c 100644 --- a/velox/exec/tests/PlanNodeSerdeTest.cpp +++ b/velox/exec/tests/PlanNodeSerdeTest.cpp @@ -1246,6 +1246,73 @@ TEST_F(PlanNodeSerdeTest, fixedPointFibonacci) { testSerde(plan); } +// stopWhenDeltaEmpty lives on the config rather than in a plan, so nothing +// else round-trips it; dropping it would silently turn a converging loop into +// one bounded only by maxIterations. +TEST_F(PlanNodeSerdeTest, fixedPointWhenDeltaEmpty) { + auto idGenerator = std::make_shared(); + auto schema = ROW({"key", "val"}, BIGINT()); + auto seed = + PlanBuilder(idGenerator) + .values({makeRowVector( + {"key", "val"}, + {makeFlatVector({0}), makeFlatVector({0})})}) + .planNode(); + auto body = PlanBuilder(idGenerator) + .stateSource("vals", schema) + .project({"key", "val + 1 AS val"}) + .planNode(); + auto plan = PlanBuilder(idGenerator) + .fixedPoint( + {core::VectorState("vals", schema).initial(seed)}, + {body}, + core::ConvergenceConfig::whenDeltaEmpty(100), + "vals") + .planNode(); + testSerde(plan); +} + +// Shape (5): a convergence criterion that is itself a shuffling sequence -- +// the only serde case where convergenceConfig carries more than one plan. +// Every other fixed point test round-trips a single convergence plan. +TEST_F(PlanNodeSerdeTest, fixedPointConvergenceSequence) { + auto idGenerator = std::make_shared(); + auto schema = ROW({"key", "val"}, BIGINT()); + auto shuffleType = ROW({"partial"}, BIGINT()); + auto seed = + PlanBuilder(idGenerator) + .values({makeRowVector( + {"key", "val"}, + {makeFlatVector({0}), makeFlatVector({0})})}) + .planNode(); + auto body = PlanBuilder(idGenerator) + .stateSource("vals", schema) + .project({"key", "val + 1 AS val"}) + .planNode(); + // Each worker reduces its shard, then every worker sums the peers' partials + // and emits the same verdict. + auto partial = PlanBuilder(idGenerator) + .stateSource("vals", schema) + .singleAggregation({}, {"sum(val)"}) + .project({"a0 AS partial"}) + .partitionedOutput({}, 2) + .planNode(); + auto verdict = PlanBuilder(idGenerator) + .exchange(shuffleType, "Presto") + .singleAggregation({}, {"sum(partial)"}) + .project({"a0 >= 16 AS converged"}) + .planNode(); + auto plan = + PlanBuilder(idGenerator) + .fixedPoint( + {core::VectorState("vals", schema).initial(seed)}, + {body}, + core::ConvergenceConfig::converging({partial, verdict}, 100), + "vals") + .planNode(); + testSerde(plan); +} + TEST_F(PlanNodeSerdeTest, fixedPointThreeDegrees) { // Shape (3): people within 3 degrees in the social graph, modeled as // hash-table reuse -- a HashTable state (the graph) built once and probed via @@ -1304,7 +1371,7 @@ TEST_F(PlanNodeSerdeTest, fixedPointThreeDegrees) { ASSERT_NE(fixedPoint, nullptr); EXPECT_EQ(fixedPoint->outputStateEntry(), "reach"); ASSERT_EQ(fixedPoint->stateDeclarations().size(), 2); - ASSERT_NE(fixedPoint->convergenceConfig().plan, nullptr); + ASSERT_EQ(fixedPoint->convergenceConfig().plans.size(), 1); const auto& hashTableState = dynamic_cast( diff --git a/velox/exec/tests/PlanNodeToStringTest.cpp b/velox/exec/tests/PlanNodeToStringTest.cpp index f8c0f36db8e..fa0a58e537b 100644 --- a/velox/exec/tests/PlanNodeToStringTest.cpp +++ b/velox/exec/tests/PlanNodeToStringTest.cpp @@ -1252,7 +1252,7 @@ TEST_F(PlanNodeToStringTest, fixedPointSequence) { ASSERT_EQ("-- FixedPoint[3]\n", plan->toString()); ASSERT_EQ( - "-- FixedPoint[3][outputStateEntry: n, states: [n (vector, append, initialized)], plans: 1, maxIterations: 9, errorWhenMaxIterationReached: false, convergencePlan: none] -> x:BIGINT\n", + "-- FixedPoint[3][outputStateEntry: n, states: [n (vector, append, initialized)], plans: 1, maxIterations: 9, errorWhenMaxIterationReached: false, stopWhenDeltaEmpty: false, convergencePlans: 0] -> x:BIGINT\n", plan->toString(true, false)); } @@ -1283,7 +1283,7 @@ TEST_F(PlanNodeToStringTest, fixedPointFibonacci) { ASSERT_EQ("-- FixedPoint[5]\n", plan->toString()); ASSERT_EQ( - "-- FixedPoint[5][outputStateEntry: fib, states: [fib (vector, replace, initialized)], plans: 1, maxIterations: 100, errorWhenMaxIterationReached: true, convergencePlan: present] -> a:BIGINT, b:BIGINT\n", + "-- FixedPoint[5][outputStateEntry: fib, states: [fib (vector, replace, initialized)], plans: 1, maxIterations: 100, errorWhenMaxIterationReached: true, stopWhenDeltaEmpty: false, convergencePlans: 1] -> a:BIGINT, b:BIGINT\n", plan->toString(true, false)); } @@ -1333,7 +1333,7 @@ TEST_F(PlanNodeToStringTest, fixedPointThreeDegrees) { ASSERT_EQ("-- FixedPoint[8]\n", plan->toString()); ASSERT_EQ( - "-- FixedPoint[8][outputStateEntry: reach, states: [graph (hashTable, keys: [src], initialized), reach (vector, append, initialized)], plans: 1, maxIterations: 3, errorWhenMaxIterationReached: true, convergencePlan: present] -> id:BIGINT, depth:BIGINT\n", + "-- FixedPoint[8][outputStateEntry: reach, states: [graph (hashTable, keys: [src], initialized), reach (vector, append, initialized)], plans: 1, maxIterations: 3, errorWhenMaxIterationReached: true, stopWhenDeltaEmpty: false, convergencePlans: 1] -> id:BIGINT, depth:BIGINT\n", plan->toString(true, false)); } @@ -1370,7 +1370,7 @@ TEST_F(PlanNodeToStringTest, fixedPointDistributed) { ASSERT_EQ("-- FixedPoint[6]\n", plan->toString()); ASSERT_EQ( - "-- FixedPoint[6][outputStateEntry: frontier, states: [frontier (vector, replace, initialized)], plans: 2, maxIterations: 3, errorWhenMaxIterationReached: false, convergencePlan: none] -> key:BIGINT, val:BIGINT\n", + "-- FixedPoint[6][outputStateEntry: frontier, states: [frontier (vector, replace, initialized)], plans: 2, maxIterations: 3, errorWhenMaxIterationReached: false, stopWhenDeltaEmpty: false, convergencePlans: 0] -> key:BIGINT, val:BIGINT\n", plan->toString(true, false)); }