Hydrator

Overview

The Hydrator trait defines how candidates are enriched with additional data. Hydrators run in parallel and populate specific fields on candidate objects by making async calls to external services.

Hydrators MUST maintain the same candidate order and count. They cannot drop or reorder candidates - use Filter stages for removal.

Trait Definition

pub trait Hydrator<Q, C>: Any + Send + Sync
where
    Q: Clone + Send + Sync + 'static,
    C: Clone + Send + Sync + 'static,
{
    fn enable(&self, query: &Q) -> bool;
    async fn hydrate(&self, query: &Q, candidates: &[C]) -> Result<Vec<C>, String>;
    fn update(&self, candidate: &mut C, hydrated: C);
    fn update_all(&self, candidates: &mut [C], hydrated: Vec<C>);
    fn name(&self) -> &'static str;
}

Type Parameters

generic

The query type that contains request context and parametersConstraints: Clone + Send + Sync + 'static

generic

The candidate type that will be hydrated with additional fieldsConstraints: Clone + Send + Sync + 'static

Methods

enable

Determines whether this hydrator should run for the given query

fn enable(&self, query: &Q) -> bool

Reference to the query object

bool

Returns true if this hydrator should run, false to skip. Default implementation returns true.

hydrate

async fn

Performs async operations to fetch data and returns candidates with hydrated fields

async fn hydrate(&self, query: &Q, candidates: &[C]) -> Result<Vec<C>, String>

Reference to the query object

&[C]

Slice of candidates to hydrate

Result<Vec<C>, String>

Returns a vector with the same length and order as input, with only this hydrator’s fields populated. Returns an error message on failure.

The returned vector MUST have the same candidates in the same order as the input. Dropping or reordering candidates is not allowed.

update

Copies hydrated fields from a hydrated candidate into the target candidate

fn update(&self, candidate: &mut C, hydrated: C)

&mut C

Mutable reference to the candidate to update

The hydrated candidate containing new field values

Only copy the fields that this hydrator is responsible for. Leave other fields unchanged.

update_all

Updates all candidates with their corresponding hydrated fields

fn update_all(&self, candidates: &mut [C], hydrated: Vec<C>)

&mut [C]

Mutable slice of candidates to update

Vec<C>

Vector of hydrated candidates from the hydrate method

The default implementation iterates and calls update for each pair. Override for custom batching logic.

name

Returns a stable name for logging and metrics

fn name(&self) -> &'static str

&'static str

A short type name derived from the implementing struct

Example Implementation

Here’s a real example that hydrates tweet author information from the Gizmoduck user service:

use xai_candidate_pipeline::hydrator::Hydrator;
use tonic::async_trait;

pub struct GizmoduckCandidateHydrator {
    pub gizmoduck_client: Arc<dyn GizmoduckClient + Send + Sync>,
}

#[async_trait]
impl Hydrator<ScoredPostsQuery, PostCandidate> for GizmoduckCandidateHydrator {
    async fn hydrate(
        &self,
        _query: &ScoredPostsQuery,
        candidates: &[PostCandidate],
    ) -> Result<Vec<PostCandidate>, String> {
        // Collect unique user IDs to fetch
        let author_ids: Vec<_> = candidates
            .iter()
            .map(|c| c.author_id as i64)
            .collect();

        // Batch fetch user data
        let users = self.gizmoduck_client
            .get_users(author_ids)
            .await
            .map_err(|e| e.to_string())?;

        // Build hydrated candidates with only the fields we populate
        let mut hydrated_candidates = Vec::with_capacity(candidates.len());
        for candidate in candidates {
            let user = users.get(&(candidate.author_id as i64));
            let author_followers_count = user
                .and_then(|u| u.counts.as_ref())
                .map(|c| c.followers_count as i32);
            let author_screen_name = user
                .and_then(|u| u.profile.as_ref())
                .map(|p| p.screen_name.clone());

            hydrated_candidates.push(PostCandidate {
                author_followers_count,
                author_screen_name,
                ..Default::default()
            });
        }

        Ok(hydrated_candidates)
    }

    fn update(&self, candidate: &mut PostCandidate, hydrated: PostCandidate) {
        // Only update the fields this hydrator is responsible for
        candidate.author_followers_count = hydrated.author_followers_count;
        candidate.author_screen_name = hydrated.author_screen_name;
    }
}

Usage Notes

Multiple hydrators run in parallel to minimize latency
Batch operations in hydrate for efficiency (e.g., fetch all users in one request)
Only populate fields relevant to this hydrator in the returned candidates
Use Default::default() for other fields to avoid accidentally overwriting data
The pipeline merges results by calling update for each hydrator
Consider implementing custom update_all for optimized batch updates

Best Practices

Maintain Order: Always return candidates in the same order as received
Batch Requests: Deduplicate IDs and make bulk requests to external services
Partial Hydration: Handle missing data gracefully with Option types
Error Handling: Provide descriptive error messages for debugging
Field Isolation: Only update fields owned by this hydrator in the update method

The QueryHydrator trait is used for hydrating the query/request object itself before candidate retrieval, rather than hydrating candidates.

pub trait QueryHydrator<Q>: Any + Send + Sync
where
    Q: Clone + Send + Sync + 'static,
{
    fn enable(&self, query: &Q) -> bool;
    async fn hydrate(&self, query: &Q) -> Result<Q, String>;
    fn update(&self, query: &mut Q, hydrated: Q);
    fn name(&self) -> &'static str;
}

Key Differences from Hydrator

Purpose

QueryHydrator: Enriches the request/query with user context (e.g., user action sequences, feature flags)
Hydrator: Enriches candidates with additional metadata (e.g., author profiles, media info)

Execution Timing

QueryHydrator: Runs early in the pipeline, before candidate sourcing
Hydrator: Runs after candidates are retrieved from sources

Type Parameters

QueryHydrator: Only generic over query type Q
Hydrator: Generic over both query Q and candidate C types

Example: UserActionSeqQueryHydrator

home-mixer/query_hydrators/user_action_seq_query_hydrator.rs

// Hydrates the query with user's recent engagement history
impl QueryHydrator<HomeMixerQuery> for UserActionSeqQueryHydrator {
    fn enable(&self, _query: &HomeMixerQuery) -> bool {
        true  // Always fetch user action sequence
    }
    
    async fn hydrate(&self, query: &HomeMixerQuery) -> Result<HomeMixerQuery, String> {
        let user_id = query.user_id;
        let actions = self.client.get_user_actions(user_id, self.max_actions).await?;
        
        let mut hydrated = query.clone();
        hydrated.user_action_sequence = Some(actions);
        Ok(hydrated)
    }
    
    fn update(&self, query: &mut HomeMixerQuery, hydrated: HomeMixerQuery) {
        query.user_action_sequence = hydrated.user_action_sequence;
    }
}

Query hydrators are essential for Phoenix models, which require user engagement history to make accurate predictions.

Candidate Pipeline Traits

Phoenix Models

Home Mixer Components

Overview

Trait Definition

Type Parameters

Methods

enable

hydrate

update

update_all

name

Example Implementation

Usage Notes

Best Practices

Key Differences from Hydrator

Example: UserActionSeqQueryHydrator

See Also

​Overview

​Trait Definition

​Type Parameters

​Methods

​enable

​hydrate

​update

​update_all

​name

​Example Implementation

​Usage Notes

​Best Practices

​Related: QueryHydrator

​Key Differences from Hydrator

​Example: UserActionSeqQueryHydrator

​See Also

Overview

Trait Definition

Type Parameters

Methods

enable

hydrate

update

update_all

name

Example Implementation

Usage Notes

Best Practices

Related: QueryHydrator

Key Differences from Hydrator

Example: UserActionSeqQueryHydrator

See Also