dbsp 0.325.0

//! Operators to shard batches across multiple worker threads based on keys
//! and to gather sharded batches in one worker.

// TODOs:
// - different sharding modes.

use feldera_storage::fbuf::{FBuf, FBufSerializer};
use itertools::Itertools;
use rkyv::{archived_root, ser::Serializer as _};

use crate::{
    Circuit, Runtime, Stream,
    circuit::{
        circuit_builder::StreamId,
        runtime::{WorkerLocation, WorkerLocations},
    },
    circuit_cache_key,
    dynamic::{Data, DataTrait, DynPair, DynPairs, Factory},
    operator::communication::{ExchangeActivity, Mailbox, new_exchange_operators},
    storage::file::SerializerInner,
    trace::{
        Batch, BatchReader, Builder, IndexedWSetSerializer, deserialize_indexed_wset, merge_batches,
    },
};

use std::{
    ops::{Not as _, Range},
    panic::Location,
};

circuit_cache_key!(ShardId<C, D>((StreamId, Range<usize>) => Stream<C, D>));
circuit_cache_key!(UnshardId<C, D>(StreamId => Stream<C, D>));

fn all_workers() -> Range<usize> {
    0..Runtime::num_workers()
}

/// Builder for sharding a batch or pairs.
pub struct Sharder<'a, C, B> {
    stream: &'a Stream<C, B>,
    workers: Range<usize>,
    activity: ExchangeActivity,
}

impl<'a, C, B> Sharder<'a, C, B>
where
    C: Circuit,
{
    /// Sets `workers` as the range of workers to shard to.  This is useful for
    /// sharding to workers on one particular host in a multihost pipeline.  By
    /// default, sharding happens across all workers.
    pub fn with_workers(self, workers: Range<usize>) -> Self {
        Self { workers, ..self }
    }

    /// Returns the sharder configured to be active during `activity`.  The
    /// default is [ExchangeActivity::AllSteps]
    pub fn with_activity(self, activity: ExchangeActivity) -> Self {
        Self { activity, ..self }
    }

    /// Shards the stream with output to a batch of the same kind.
    #[track_caller]
    pub fn shard(self, factories: &B::Factories) -> Stream<C, B>
    where
        B: Batch<Time = ()> + Clone + Send,
    {
        self.shard_generic_ref(factories)
            .unwrap_or_else(|| self.stream.clone())
    }
}

impl<'a, C, IB> Sharder<'a, C, IB>
where
    C: Circuit,
    IB: BatchReader<Time = ()> + Clone,
{
    /// Shards the stream with output to batch type `OB`.  If the stream doesn't
    /// need to be sharded because there is only a single worker, returns
    /// `None`.
    #[track_caller]
    pub fn shard_generic<OB>(self, factories: &OB::Factories) -> Option<Stream<C, OB>>
    where
        OB: Batch<Key = IB::Key, Val = IB::Val, Time = (), R = IB::R> + Send,
    {
        self.shard_generic_ref(factories)
    }

    #[track_caller]
    fn shard_generic_ref<OB>(&self, factories: &OB::Factories) -> Option<Stream<C, OB>>
    where
        OB: Batch<Key = IB::Key, Val = IB::Val, Time = (), R = IB::R> + Send,
    {
        if Runtime::num_workers() == 1 {
            return None;
        }
        let location = Location::caller();
        let output = self
            .stream
            .circuit()
            .cache_get_or_insert_with(
                ShardId::new((self.stream.stream_id(), self.workers.clone())),
                move || {
                    // As a minor optimization, we reuse this array across all invocations
                    // of the sharding operator.
                    let mut builders = Vec::with_capacity(Runtime::num_workers());
                    let factories_clone2 = factories.clone();
                    let factories_clone3 = factories.clone();
                    let factories_clone4 = factories.clone();
                    let workers_clone = self.workers.clone();
                    let workers_clone2 = self.workers.clone();

                    let output = self.stream.circuit().region("shard", || {
                        let (sender, receiver) = new_exchange_operators(
                            Some(location),
                            || Vec::new(),
                            move |batch: IB, batches: &mut Vec<Mailbox<OB>>| {
                                shard_batch(
                                    batch,
                                    &workers_clone,
                                    &mut builders,
                                    batches,
                                    &factories_clone3,
                                );
                            },
                            move |data| deserialize_indexed_wset(&factories_clone4, &data),
                            |batches: &mut Vec<OB>, batch: OB| batches.push(batch),
                            self.activity,
                        )
                        .unwrap();

                        self.stream
                            .circuit()
                            .add_exchange(sender, receiver, self.stream)
                            .apply_owned_named("merge shards", move |batches| {
                                merge_batches(&factories_clone2, batches, &None, &None)
                            })
                    });

                    self.stream.circuit().cache_insert(
                        ShardId::new((output.stream_id(), workers_clone2)),
                        output.clone(),
                    );

                    self.stream
                        .circuit()
                        .cache_insert(UnshardId::new(output.stream_id()), self.stream.clone());

                    output.set_persistent_id(
                        self.stream
                            .get_persistent_id()
                            .map(|name| format!("{name}.shard"))
                            .as_deref(),
                    )
                },
            )
            .clone();

        Some(output)
    }
}

impl<'a, C, K, V> Sharder<'a, C, Vec<Box<DynPairs<K, V>>>>
where
    C: Circuit,
    K: DataTrait + ?Sized,
    V: DataTrait + ?Sized,
{
    /// Shards a vector of pairs across workers.
    #[track_caller]
    pub fn shard_pairs(
        self,
        pairs_factory: &'static dyn Factory<DynPairs<K, V>>,
    ) -> Stream<C, Vec<Box<DynPairs<K, V>>>> {
        if self.stream.is_sharded() {
            return self.stream.clone();
        }

        let location = Location::caller();

        let (sender, receiver) = new_exchange_operators(
            Some(location),
            Vec::new,
            move |input_pairs: Vec<Box<DynPairs<K, V>>>,
                  output_pairs: &mut Vec<Mailbox<Box<DynPairs<K, V>>>>| {
                shard_pairs(input_pairs, &self.workers, output_pairs, pairs_factory);
            },
            move |data| deserialize_pairs(&data, pairs_factory),
            |output_pairs: &mut Vec<Box<DynPairs<K, V>>>, batch: Box<DynPairs<K, V>>| {
                output_pairs.push(batch);
            },
            self.activity,
        )
        .unwrap();

        let output = self
            .stream
            .circuit()
            .add_exchange(sender, receiver, self.stream);

        output.set_persistent_id(
            self.stream
                .get_persistent_id()
                .map(|name| format!("{name}.shard"))
                .as_deref(),
        );
        output
    }
}

impl<C, B> Stream<C, B> {
    /// Constructs a [Sharder] builder for sharding a batch or pairs with
    /// flexible configuration.
    pub fn dyn_sharder(&self) -> Sharder<'_, C, B> {
        Sharder {
            stream: self,
            activity: ExchangeActivity::AllSteps,
            workers: all_workers(),
        }
    }
}

impl<C, IB> Stream<C, IB>
where
    C: Circuit,
    IB: BatchReader<Time = ()> + Clone,
{
    /// See [`Stream::shard`].
    #[track_caller]
    pub fn dyn_shard(&self, factories: &IB::Factories) -> Stream<C, IB>
    where
        IB: Batch + Send,
    {
        self.dyn_sharder().shard(factories)
    }

    /// See [`Stream::shard_workers`].
    #[track_caller]
    pub fn dyn_shard_workers(
        &self,
        workers: Range<usize>,
        factories: &IB::Factories,
    ) -> Stream<C, IB>
    where
        IB: Batch + Send,
    {
        self.dyn_sharder().with_workers(workers).shard(factories)
    }

    /// Like [`Self::dyn_shard`], but can assemble the results into any output batch
    /// type `OB`.
    ///
    /// Returns `None` when the circuit is not running inside a multithreaded
    /// runtime or is running in a runtime with a single worker thread.
    #[track_caller]
    pub fn dyn_shard_generic<OB>(&self, factories: &OB::Factories) -> Option<Stream<C, OB>>
    where
        OB: Batch<Key = IB::Key, Val = IB::Val, Time = (), R = IB::R> + Send,
    {
        self.dyn_sharder().shard_generic(factories)
    }

    /// Like [`Self::dyn_shard`], but can assemble the results into any output batch
    /// type `OB`.
    ///
    /// Returns `None` when the circuit is not running inside a multithreaded
    /// runtime or is running in a runtime with a single worker thread.
    #[track_caller]
    pub fn dyn_shard_generic_workers<OB>(
        &self,
        workers: Range<usize>,
        factories: &OB::Factories,
    ) -> Option<Stream<C, OB>>
    where
        OB: Batch<Key = IB::Key, Val = IB::Val, Time = (), R = IB::R> + Send,
    {
        self.dyn_sharder()
            .with_workers(workers)
            .shard_generic(factories)
    }
}

impl<C, K, V> Stream<C, Vec<Box<DynPairs<K, V>>>>
where
    C: Circuit,
    K: DataTrait + ?Sized,
    V: DataTrait + ?Sized,
{
    #[track_caller]
    pub fn dyn_shard_pairs(
        &self,
        pairs_factory: &'static dyn Factory<DynPairs<K, V>>,
    ) -> Stream<C, Vec<Box<DynPairs<K, V>>>> {
        self.dyn_sharder().shard_pairs(pairs_factory)
    }
}

fn deserialize_pairs<K, V>(
    data: &[u8],
    pairs_factory: &'static dyn Factory<DynPairs<K, V>>,
) -> Box<DynPairs<K, V>>
where
    K: DataTrait + ?Sized,
    V: DataTrait + ?Sized,
{
    let offsets = unsafe { archived_root::<Vec<usize>>(data) };
    let mut output = pairs_factory.default_box();
    output.reserve(offsets.len());
    for offset in (0..offsets.len()).map(|i| offsets[i] as usize) {
        output.push_with(&mut |pair| {
            unsafe { pair.deserialize_from_bytes(data, offset) };
        })
    }
    output
}

pub(crate) enum ShardBuilder<OB>
where
    OB: Batch<Time = ()>,
{
    Local(OB::Builder),
    Remote(IndexedWSetSerializer),
}

impl<OB> ShardBuilder<OB>
where
    OB: Batch<Time = ()>,
{
    fn new(
        location: WorkerLocation,
        factories: &OB::Factories,
        estimated_keys: usize,
        estimated_values: usize,
    ) -> Self {
        match location {
            WorkerLocation::Local => Self::Local(OB::Builder::with_capacity(
                factories,
                estimated_keys,
                estimated_values,
            )),
            WorkerLocation::Remote => Self::Remote(IndexedWSetSerializer::with_capacity(
                estimated_keys,
                estimated_values,
            )),
        }
    }

    fn push_diff(&mut self, weight: &OB::R, serializer_inner: &mut Option<SerializerInner>) {
        match self {
            ShardBuilder::Local(builder) => builder.push_diff(weight),
            ShardBuilder::Remote(serializer) => {
                serializer.push_diff(weight, serializer_inner.get_or_insert_default())
            }
        }
    }

    fn push_diff_mut(
        &mut self,
        weight: &mut OB::R,
        serializer_inner: &mut Option<SerializerInner>,
    ) {
        match self {
            ShardBuilder::Local(builder) => builder.push_diff_mut(weight),
            ShardBuilder::Remote(serializer) => {
                serializer.push_diff(weight, serializer_inner.get_or_insert_default())
            }
        }
    }

    fn push_val(&mut self, val: &OB::Val, serializer_inner: &mut Option<SerializerInner>) {
        match self {
            ShardBuilder::Local(builder) => builder.push_val(val),
            ShardBuilder::Remote(serializer) => {
                serializer.push_val(val, serializer_inner.get_or_insert_default())
            }
        }
    }

    fn push_val_mut(&mut self, val: &mut OB::Val, serializer_inner: &mut Option<SerializerInner>) {
        match self {
            ShardBuilder::Local(builder) => builder.push_val_mut(val),
            ShardBuilder::Remote(serializer) => {
                serializer.push_val(val, serializer_inner.get_or_insert_default())
            }
        }
    }

    fn push_key(&mut self, key: &OB::Key, serializer_inner: &mut Option<SerializerInner>) {
        match self {
            ShardBuilder::Local(builder) => builder.push_key(key),
            ShardBuilder::Remote(serializer) => {
                serializer.push_key(key, serializer_inner.get_or_insert_default())
            }
        }
    }

    fn push_key_mut(&mut self, key: &mut OB::Key, serializer_inner: &mut Option<SerializerInner>) {
        match self {
            ShardBuilder::Local(builder) => builder.push_key_mut(key),
            ShardBuilder::Remote(serializer) => {
                serializer.push_key(key, serializer_inner.get_or_insert_default())
            }
        }
    }

    fn done(self, serializer_inner: &mut Option<SerializerInner>) -> Mailbox<OB> {
        match self {
            ShardBuilder::Local(builder) => Mailbox::Plain(builder.done()),
            ShardBuilder::Remote(serializer) => {
                Mailbox::Tx(serializer.done(serializer_inner.get_or_insert_default()))
            }
        }
    }
}

// Partitions the batch into shards covering `workers` (out of
// `all_workers()`), based on the hash of the key.
pub(crate) fn shard_batch<IB, OB>(
    mut batch: IB,
    workers: &Range<usize>,
    builders: &mut Vec<ShardBuilder<OB>>,
    outputs: &mut Vec<Mailbox<OB>>,
    factories: &OB::Factories,
) where
    IB: BatchReader<Time = ()>,
    OB: Batch<Key = IB::Key, Val = IB::Val, Time = (), R = IB::R>,
{
    builders.clear();

    // XXX If `shards == 1` and `OB` and `IB` are the same, then we could
    // implement this more efficiently, without copying.
    let shards = workers.len();
    let keys_per_shard = batch.key_count() / shards;
    let values_per_shard = batch.len() / shards;
    for (worker, location) in WorkerLocations::new().enumerate() {
        let (estimated_keys, estimated_values) = if workers.contains(&worker) {
            (keys_per_shard, values_per_shard)
        } else {
            (0, 0)
        };
        builders.push(ShardBuilder::new(
            location,
            factories,
            estimated_keys,
            estimated_values,
        ));
    }

    let mut serializer_inner = None;
    let mut cursor = batch.consuming_cursor(None, None);
    if cursor.has_mut() {
        while cursor.key_valid() {
            let b = &mut builders[cursor.key().default_hash() as usize % shards + workers.start];
            while cursor.val_valid() {
                b.push_diff_mut(cursor.weight_mut(), &mut serializer_inner);
                b.push_val_mut(cursor.val_mut(), &mut serializer_inner);
                cursor.step_val();
            }
            b.push_key_mut(cursor.key_mut(), &mut serializer_inner);
            cursor.step_key();
        }
    } else {
        while cursor.key_valid() {
            let b = &mut builders[cursor.key().default_hash() as usize % shards + workers.start];
            while cursor.val_valid() {
                b.push_diff(cursor.weight(), &mut serializer_inner);
                b.push_val(cursor.val(), &mut serializer_inner);
                cursor.step_val();
            }
            b.push_key(cursor.key(), &mut serializer_inner);
            cursor.step_key();
        }
    }
    for builder in builders.drain(..) {
        outputs.push(builder.done(&mut serializer_inner));
    }
}

pub struct PairsSerializer {
    fbuf: FBuf,
    offsets: Vec<usize>,
}

impl PairsSerializer {
    pub fn with_capacity(estimated_pairs: usize) -> Self {
        Self {
            fbuf: FBuf::default(),
            offsets: Vec::with_capacity(estimated_pairs),
        }
    }

    pub fn push_val<K, V>(&mut self, pair: &DynPair<K, V>, serializer: &mut SerializerInner)
    where
        K: DataTrait + ?Sized,
        V: DataTrait + ?Sized,
    {
        self.offsets.push(
            serializer
                .with(FBufSerializer::new(&mut self.fbuf), |s| pair.serialize(s))
                .unwrap(),
        );
    }

    pub fn done(mut self, serializer: &mut SerializerInner) -> FBuf {
        serializer
            .with(FBufSerializer::new(&mut self.fbuf), |s| {
                s.serialize_value(&self.offsets)
            })
            .unwrap();
        self.fbuf
    }
}

enum PairsBuilder<K, V>
where
    K: DataTrait + ?Sized,
    V: DataTrait + ?Sized,
{
    Local(Box<DynPairs<K, V>>),
    Remote(PairsSerializer),
}

impl<K, V> PairsBuilder<K, V>
where
    K: DataTrait + ?Sized,
    V: DataTrait + ?Sized,
{
    fn with_capacity(
        location: WorkerLocation,
        pairs_factory: &'static dyn Factory<DynPairs<K, V>>,
        estimated_pairs: usize,
    ) -> Self {
        match location {
            WorkerLocation::Local => {
                let mut pairs = pairs_factory.default_box();
                pairs.reserve(estimated_pairs);
                Self::Local(pairs)
            }
            WorkerLocation::Remote => Self::Remote(PairsSerializer::with_capacity(estimated_pairs)),
        }
    }

    fn push_val(&mut self, pair: &mut DynPair<K, V>, inner: &mut Option<SerializerInner>) {
        match self {
            PairsBuilder::Local(pairs) => pairs.push_val(pair),
            PairsBuilder::Remote(serializer) => {
                serializer.push_val(pair, inner.get_or_insert_default())
            }
        }
    }

    fn done(self, inner: &mut Option<SerializerInner>) -> Mailbox<Box<DynPairs<K, V>>> {
        match self {
            PairsBuilder::Local(pairs) => Mailbox::Plain(pairs),
            PairsBuilder::Remote(serializer) => {
                Mailbox::Tx(serializer.done(inner.get_or_insert_default()))
            }
        }
    }
}

// Partitions the batch into shards covering `workers` (out of
// `all_workers()`), based on the hash of the key.
pub fn shard_pairs<K, V>(
    input_pairs: Vec<Box<DynPairs<K, V>>>,
    workers: &Range<usize>,
    output_pairs: &mut Vec<Mailbox<Box<DynPairs<K, V>>>>,
    pairs_factory: &'static dyn Factory<DynPairs<K, V>>,
) where
    K: DataTrait + ?Sized,
    V: DataTrait + ?Sized,
{
    let pairs_per_shard =
        input_pairs.iter().map(|pairs| pairs.len()).sum::<usize>() / workers.len();
    let mut serializer_inner = None;
    let mut output = WorkerLocations::new()
        .enumerate()
        .map(|(worker, location)| {
            let estimated_pairs = if workers.contains(&worker) {
                pairs_per_shard
            } else {
                0
            };
            PairsBuilder::with_capacity(location, pairs_factory, estimated_pairs)
        })
        .collect_vec();

    // Merge the inputs and shard them into the outputs.
    let mut inputs = input_pairs
        .into_iter()
        .flat_map(|pairs| pairs.is_empty().not().then_some((pairs, 0)))
        .collect_vec();
    while let Some(min_index) = inputs
        .iter()
        .map(|(pairs, index)| pairs.index(*index))
        .position_min()
    {
        let (pairs, pairs_index) = &mut inputs[min_index];
        let pair = &mut pairs[*pairs_index];
        let shard_index = pair.fst().default_hash() as usize % workers.len() + workers.start;
        output[shard_index].push_val(pair, &mut serializer_inner);

        *pairs_index += 1;
        if *pairs_index >= pairs.len() {
            inputs.remove(min_index);
        }
    }
    output_pairs.extend(
        output
            .into_iter()
            .map(|pairs| pairs.done(&mut serializer_inner)),
    );
}

impl<C, T> Stream<C, T>
where
    C: Circuit,
    T: 'static,
{
    /// Marks the data within the current stream as sharded, meaning that all
    /// further calls to `.shard()` will have no effect.
    ///
    /// This must only be used on streams of values that are properly sharded
    /// across workers, otherwise this will cause the dataflow to yield
    /// incorrect results
    pub fn mark_sharded(&self) -> Self {
        self.mark_sharded_workers(all_workers())
    }

    pub fn mark_sharded_workers(&self, workers: Range<usize>) -> Self {
        self.circuit()
            .cache_insert(ShardId::new((self.stream_id(), workers)), self.clone());
        self.clone()
    }

    /// Returns `true` if a sharded version of the current stream exists
    pub fn has_sharded_version(&self) -> bool {
        self.has_workers_sharded_version(all_workers())
    }

    /// Returns `true` if a version of the current stream sharded to `workers`
    /// exists.
    pub fn has_workers_sharded_version(&self, workers: Range<usize>) -> bool {
        self.circuit()
            .cache_contains(&ShardId::<C, T>::new((self.stream_id(), workers)))
    }

    pub fn get_sharded_version(&self) -> Option<Self> {
        self.circuit()
            .cache_get(&ShardId::<C, T>::new((self.stream_id(), all_workers())))
    }

    /// Returns the sharded version of the stream if it exists
    /// (which may be the stream itself or the result of applying
    /// the `shard` operator to it).  Otherwise, returns `self`.
    pub fn try_sharded_version(&self) -> Self {
        self.get_sharded_version().unwrap_or_else(|| self.clone())
    }

    /// Returns the unsharded version of the stream if it exists, and otherwise
    /// `self`.
    pub fn try_unsharded_version(&self) -> Self {
        self.circuit()
            .cache_get(&UnshardId::new(self.stream_id()))
            .unwrap_or_else(|| self.clone())
    }

    /// Returns `true` if this stream is sharded.
    pub fn is_sharded(&self) -> bool {
        if Runtime::num_workers() == 1 {
            return true;
        }

        self.circuit()
            .cache_get(&ShardId::<C, T>::new((self.stream_id(), all_workers())))
            .is_some_and(|sharded| sharded.ptr_eq(self))
    }

    /// Marks `self` as sharded if `input` has a sharded version of itself
    pub fn mark_sharded_if<C2, U>(&self, input: &Stream<C2, U>)
    where
        C2: Circuit,
        U: 'static,
    {
        if input.has_sharded_version() {
            self.mark_sharded();
        }
    }
}

#[cfg(test)]
mod tests {
    use crate::{
        Circuit, RootCircuit, Runtime, operator::Generator, trace::BatchReader,
        typed_batch::OrdIndexedZSet, utils::Tup2,
    };

    #[test]
    fn test_shard() {
        do_test_shard(2);
        do_test_shard(4);
        do_test_shard(16);
    }

    fn test_data(worker_index: usize, num_workers: usize) -> OrdIndexedZSet<u64, u64> {
        let tuples: Vec<_> = (0..1000)
            .filter(|n| n % num_workers == worker_index)
            .flat_map(|n| {
                vec![
                    Tup2(Tup2(n as u64, n as u64), 1i64),
                    Tup2(Tup2(n as u64, 1000 * n as u64), 1),
                ]
            })
            .collect();
        <OrdIndexedZSet<u64, u64>>::from_tuples((), tuples)
    }

    fn do_test_shard(workers: usize) {
        let hruntime = Runtime::run(workers, |_parker| {
            let circuit = RootCircuit::build(move |circuit| {
                let input = circuit.add_source(Generator::new(|| {
                    let worker_index = Runtime::worker_index();
                    let num_workers = Runtime::num_workers();
                    test_data(worker_index, num_workers)
                }));
                input
                    .shard()
                    .gather(0)
                    .inspect(|batch: &OrdIndexedZSet<u64, u64>| {
                        if Runtime::worker_index() == 0 {
                            assert_eq!(batch, &test_data(0, 1))
                        } else {
                            assert_eq!(batch.len(), 0);
                        }
                    });
                Ok(())
            })
            .unwrap()
            .0;

            for _ in 0..3 {
                circuit.transaction().unwrap();
            }
        })
        .expect("failed to run runtime");

        hruntime.join().unwrap();
    }
}