[presto][rift] Add riftTier field to RPCNode for coordinator-to-worker tier injection (#27725)

[abhinavmuk04]

Summary:

Adds a riftTier field to the RPCNode plan node across all layers (Java →
C++ protocol → Velox) so the coordinator can pass the auto-provisioned RIFT
SMC tier to workers through the query plan.

This is the follow-up to D101101436 that completes the Mode 2 (auto-start)
data path. The RIFT CLI generates random tier names (e.g.,
smc.rift_amukher1_llama3_a3f1b2m), so the coordinator must communicate the
tier through the plan node rather than deterministic naming.

Changes across the full plan node chain:

• Java RPCNode: add riftTier field, both constructors, getter, pass through
  in replaceChildren/assignStatsEquivalentPlanNode
• Java RpcFunctionOptimizer: pass empty riftTier (coordinator injection TBD)
• Java UnaliasSymbolReferences, PruneUnreferencedOutputs: pass through
• C++ presto_protocol_core: add riftTier to RPCNode struct + serialization
• C++ PrestoToVeloxQueryPlan: pass node->riftTier to Velox RPCNode
• Velox PlanNode.h/cpp: add riftTier field, getter, serialize/deserialize
• AsyncRPCFunction.h: add virtual setRiftTier() method (default no-op)
• RPCOperator.cpp: call function_->setRiftTier() before initialize()
• FbLlmInference: implement setRiftTier(), use plan node tier with highest
  priority (options JSON > plan node riftTier > session property)

Differential Revision: D101105948

[sourcery-ai]

Reviewer's Guide

Adds a riftTier string field to RPCNode and threads it end-to-end from the Presto Java planner through the C++ protocol and Velox plan nodes to async RPC functions, allowing workers (notably FbLlmInference) to receive an explicit RIFT SMC tier chosen by the coordinator.

Sequence diagram for riftTier injection from coordinator to FbLlmInference

sequenceDiagram
    actor Coordinator
    participant JavaPlanner
    participant CppProtocol as PrestoCppProtocol
    participant VeloxPlan
    participant RPCOp as RPCOperator
    participant Func as FbLlmInference
    participant RIFT as RIFTService

    Coordinator->>JavaPlanner: Build plan with RPCNode(riftTier)
    JavaPlanner->>CppProtocol: Serialize RPCNode with riftTier
    CppProtocol->>VeloxPlan: Convert to Velox RPCNode(riftTier)
    VeloxPlan->>RPCOp: Create RPCOperator with plan node riftTier
    RPCOp->>Func: setRiftTier(planNode.riftTier)
    RPCOp->>Func: initialize()
    Func->>Func: selectTier(optionsJson, planNode.riftTier, sessionProperty)
    Func->>RIFT: Invoke RPC on selected tier

Loading

Class diagram for RPCNode riftTier propagation across layers

classDiagram
    class Java_RPCNode {
        +PlanNode source
        +String functionName
        +List~RowExpression~ arguments
        +List~String~ argumentColumns
        +VariableReferenceExpression outputVariable
        +StreamingMode streamingMode
        +int dispatchBatchSize
        +String riftTier
        +Java_RPCNode(sourceLocation, id, source, functionName, arguments, argumentColumns, outputVariable, streamingMode, dispatchBatchSize, riftTier)
        +Java_RPCNode(sourceLocation, id, statsEquivalentPlanNode, source, functionName, arguments, argumentColumns, outputVariable, streamingMode, dispatchBatchSize, riftTier)
        +String getRiftTier()
        +PlanNode replaceChildren(newChildren)
        +PlanNode assignStatsEquivalentPlanNode(statsEquivalentPlanNode)
    }

    class Cpp_RPCNode {
        +PlanNode source
        +String functionName
        +vector~RowExpression~ arguments
        +vector~String~ argumentColumns
        +VariableReferenceExpression outputVariable
        +RPCNodeStreamingMode streamingMode
        +Integer dispatchBatchSize
        +String riftTier
        +RPCNode()
    }

    class Velox_RPCNode {
        +core::PlanNodePtr source
        +std::string functionName
        +std::vector~RowExpressionPtr~ arguments
        +std::vector~std::string~ argumentColumns
        +core::TypedExprPtr outputVariable
        +StreamingMode streamingMode
        +int32_t dispatchBatchSize
        +std::string riftTier
        +Velox_RPCNode(id, source, functionName, arguments, argumentColumns, outputVariable, streamingMode, dispatchBatchSize, riftTier)
        +const std::string& riftTier()
        +serialize()
        +deserialize()
    }

    class VeloxQueryPlanConverterBase {
        +core::PlanNodePtr toVeloxQueryPlan(protocol::RPCNode node, TableWriteInfo tableWriteInfo, TaskId taskId)
    }

    class AsyncRPCFunction {
        <<interface>>
        +initialize()
        +virtual void setRiftTier(const std::string& riftTier)
    }

    class FbLlmInference {
        +std::string optionsTier
        +std::string planNodeTier
        +std::string sessionTier
        +std::string effectiveTier
        +void setRiftTier(const std::string& riftTier)
        +void initialize()
        +void selectTier()
    }

    class RPCOperator {
        +std::unique_ptr~AsyncRPCFunction~ function_
        +void addInput()
        +void initialize()
        +void getOutput()
        +void prepareFunction(const std::string& riftTier)
    }

    Java_RPCNode <.. Cpp_RPCNode : json
    Cpp_RPCNode <.. Velox_RPCNode : converted_by
    VeloxQueryPlanConverterBase --> Velox_RPCNode : creates
    VeloxQueryPlanConverterBase --> Cpp_RPCNode : reads riftTier
    RPCOperator --> AsyncRPCFunction : uses
    AsyncRPCFunction <|.. FbLlmInference
    RPCOperator --> Velox_RPCNode : reads riftTier

Loading

File-Level Changes

Change	Details	Files
Add riftTier as a first-class property on Java RPCNode and propagate it through planner rewrites.	Extend RPCNode with a new riftTier field, initialize it in both constructors with null-safe default to empty string, and expose it via a @JsonProperty getter. Update replaceChildren and assignStatsEquivalentPlanNode to preserve the riftTier when cloning RPCNode instances. Modify PruneUnreferencedOutputs and UnaliasSymbolReferences visitors to pass through the existing riftTier when they rebuild RPCNode instances. Update RpcFunctionOptimizer to construct RPCNode with an empty riftTier placeholder until coordinator injection is implemented.	presto-main-base/src/main/java/com/facebook/presto/sql/planner/plan/RPCNode.java presto-main-base/src/main/java/com/facebook/presto/sql/planner/optimizations/PruneUnreferencedOutputs.java presto-main-base/src/main/java/com/facebook/presto/sql/planner/optimizations/UnaliasSymbolReferences.java presto-main-base/src/main/java/com/facebook/presto/sql/planner/optimizations/RpcFunctionOptimizer.java
Extend the Presto C++ protocol RPCNode with riftTier and convert it into Velox RPC plan nodes.	Add a riftTier String field to the C++ protocol::RPCNode definition. Update to_json and from_json for protocol::RPCNode to serialize and deserialize the riftTier field with proper metadata. Ensure VeloxQueryPlanConverterBase::toVeloxQueryPlan validates the RPC node pointer and forwards node->riftTier when constructing the Velox RPC plan node.	presto-native-execution/presto_cpp/presto_protocol/core/presto_protocol_core.h presto-native-execution/presto_cpp/presto_protocol/core/presto_protocol_core.cpp presto-native-execution/presto_cpp/main/types/PrestoToVeloxQueryPlan.cpp
Thread riftTier through Velox RPC plan nodes and into async RPC functions, with FbLlmInference honoring it in tier selection.	Extend Velox core RPC plan node representation to include a riftTier field, getter, and serialization/deserialization logic. Add a virtual setRiftTier method (default no-op) to AsyncRPCFunction and invoke it from RPCOperator before initialize() so functions can observe the tier. Implement setRiftTier in the FbLlmInference RPC function and make its tier selection prefer: options JSON tier, then plan node riftTier, then session property.	velox/core/PlanNode.h velox/core/PlanNode.cpp velox/exec/AsyncRPCFunction.h velox/exec/RPCOperator.cpp velox/exec/fb/FbLlmInference.cpp

---

Tips and commands

Interacting with Sourcery

• Trigger a new review: Comment @sourcery-ai review on the pull request.
• Continue discussions: Reply directly to Sourcery's review comments.
• Generate a GitHub issue from a review comment: Ask Sourcery to create an
  issue from a review comment by replying to it. You can also reply to a
  review comment with @sourcery-ai issue to create an issue from it.
• Generate a pull request title: Write @sourcery-ai anywhere in the pull
  request title to generate a title at any time. You can also comment
  @sourcery-ai title on the pull request to (re-)generate the title at any time.
• Generate a pull request summary: Write @sourcery-ai summary anywhere in
  the pull request body to generate a PR summary at any time exactly where you
  want it. You can also comment @sourcery-ai summary on the pull request to
  (re-)generate the summary at any time.
• Generate reviewer's guide: Comment @sourcery-ai guide on the pull
  request to (re-)generate the reviewer's guide at any time.
• Resolve all Sourcery comments: Comment @sourcery-ai resolve on the
  pull request to resolve all Sourcery comments. Useful if you've already
  addressed all the comments and don't want to see them anymore.
• Dismiss all Sourcery reviews: Comment @sourcery-ai dismiss on the pull
  request to dismiss all existing Sourcery reviews. Especially useful if you
  want to start fresh with a new review - don't forget to comment
  @sourcery-ai review to trigger a new review!

Customizing Your Experience

Access your dashboard to:

• Enable or disable review features such as the Sourcery-generated pull request
  summary, the reviewer's guide, and others.
• Change the review language.
• Add, remove or edit custom review instructions.
• Adjust other review settings.

Getting Help

• Contact our support team for questions or feedback.
• Visit our documentation for detailed guides and information.
• Keep in touch with the Sourcery team by following us on X/Twitter, LinkedIn or GitHub.

[github-actions]

Dependency Review

✅ No vulnerabilities or license issues or OpenSSF Scorecard issues found.

Scanned Files

None

[sourcery-ai]

Hey - I've left some high level feedback:

• The Java RPCNode constructor silently normalizes null riftTier values to an empty string while the C++ RPCNode just exposes a String riftTier field; consider aligning the default/empty semantics across layers (e.g., consistently treating missing vs empty tiers) to avoid subtle mismatches in behavior.
• In RpcFunctionOptimizer the new riftTier argument is hardcoded as an empty string; if this is meant to represent an unset tier, consider using a named constant or helper to make the intent clear and to avoid scattering magic values if tier injection logic changes later.

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- The Java `RPCNode` constructor silently normalizes `null` `riftTier` values to an empty string while the C++ `RPCNode` just exposes a `String riftTier` field; consider aligning the default/empty semantics across layers (e.g., consistently treating missing vs empty tiers) to avoid subtle mismatches in behavior.
- In `RpcFunctionOptimizer` the new `riftTier` argument is hardcoded as an empty string; if this is meant to represent an unset tier, consider using a named constant or helper to make the intent clear and to avoid scattering magic values if tier injection logic changes later.

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}

[steveburnett]

• Please edit the PR title to follow semantic commit style to pass the failing and required CI check. See the failure in the test for advice. If you can’t edit the PR title let us know and we can help.

• Please add a release note - or NO RELEASE NOTE - following the Release Notes Guidelines to pass the failing but not required CI check.