add global option to loosen ordering guarantees in stateless transforms in exchange for potential performance benefits

Author: ArunPiduguDD

lib/vector-core/src/config/global_options.rs

@@ -190,6 +190,15 @@ pub struct GlobalOptions {
     /// `find_vector_metrics`, and `aggregate_vector_metrics` functions.
     #[serde(default, skip_serializing_if = "crate::serde::is_default")]
     pub metrics_storage_refresh_period: Option<f64>,
+
+    /// Whether to preserve event ordering for concurrent stateless transforms.
+    ///
+    /// When `true` (the default), concurrent stateless transforms use `FuturesOrdered`,
+    /// guaranteeing output order matches input order. When `false`, `FuturesUnordered`
+    /// is used, which may improve throughput at the cost of ordering guarantees.
+    #[serde(default, skip_serializing_if = "crate::serde::is_default")]
+    #[configurable(metadata(docs::common = false, docs::required = false))]
+    pub preserve_ordering_stateless_transforms: Option<bool>,
 }
 
 impl_generate_config_from_default!(GlobalOptions);
@@ -295,6 +304,16 @@ impl GlobalOptions {
             errors.push("conflicting values for 'expire_metrics_secs' found".to_owned());
         }
 
+        if conflicts(
+            self.preserve_ordering_stateless_transforms.as_ref(),
+            with.preserve_ordering_stateless_transforms.as_ref(),
+        ) {
+            errors.push(
+                "conflicting values for 'preserve_ordering_stateless_transforms' found"
+                    .to_owned(),
+            );
+        }
+
         let data_dir = if self.data_dir.is_none() || self.data_dir == default_data_dir() {
             with.data_dir
         } else if with.data_dir != default_data_dir() && self.data_dir != with.data_dir {
@@ -345,6 +364,9 @@ impl GlobalOptions {
                 metrics_storage_refresh_period: self
                     .metrics_storage_refresh_period
                     .or(with.metrics_storage_refresh_period),
+                preserve_ordering_stateless_transforms: self
+                    .preserve_ordering_stateless_transforms
+                    .or(with.preserve_ordering_stateless_transforms),
             })
         } else {
             Err(errors)

src/topology/builder.rs

@@ -6,7 +6,7 @@ use std::{
     time::Instant,
 };
 
-use futures::{FutureExt, StreamExt, TryStreamExt, stream::FuturesOrdered};
+use futures::{FutureExt, StreamExt, TryStreamExt};
 use futures_util::stream::FuturesUnordered;
 use metrics::gauge;
 use stream_cancel::{StreamExt as StreamCancelExt, Trigger, Tripwire};
@@ -745,13 +745,19 @@ impl<'a> Builder<'a> {
         let sender = self
             .utilization_registry
             .add_component(node.key.clone(), gauge!("utilization"));
+        let preserve_ordering = self
+            .config
+            .global
+            .preserve_ordering_stateless_transforms
+            .unwrap_or(true);
         let runner = Runner::new(
             t,
             input_rx,
             sender,
             node.input_details.data_type(),
             outputs,
             LatencyRecorder::new(self.config.global.latency_ewma_alpha),
+            preserve_ordering,
         );
         let transform = if node.enable_concurrency {
             runner.run_concurrently().boxed()
@@ -1125,6 +1131,7 @@ struct Runner {
     timer_tx: UtilizationComponentSender,
     latency_recorder: LatencyRecorder,
     events_received: Registered<EventsReceived>,
+    preserve_ordering: bool,
 }
 
 impl Runner {
@@ -1135,6 +1142,7 @@ impl Runner {
         input_type: DataType,
         outputs: TransformOutputs,
         latency_recorder: LatencyRecorder,
+        preserve_ordering: bool,
     ) -> Self {
         Self {
             transform,
@@ -1144,6 +1152,7 @@ impl Runner {
             timer_tx,
             latency_recorder,
             events_received: register!(EventsReceived),
+            preserve_ordering,
         }
     }
 
@@ -1199,7 +1208,8 @@ impl Runner {
         let mut input_rx =
             super::ready_arrays::ReadyArrays::with_capacity(input_rx, READY_ARRAY_CAPACITY);
 
-        let mut in_flight = FuturesOrdered::new();
+        let mut in_flight =
+            super::in_flight_queue::InFlightQueue::new(self.preserve_ordering);
         let mut shutting_down = false;
 
         self.timer_tx.try_send_start_wait();
@@ -1234,7 +1244,7 @@ impl Runner {
                                 }
                                 outputs_buf
                             }.in_current_span());
-                            in_flight.push_back(task);
+                            in_flight.push(task);
                         }
                         None => {
                             shutting_down = true;

src/topology/in_flight_queue.rs

@@ -0,0 +1,57 @@
+use std::{future::Future, pin::Pin};
+
+use futures::{
+    Stream,
+    stream::{FuturesOrdered, FuturesUnordered},
+    task::{Context, Poll},
+};
+
+/// Wraps either a [`FuturesOrdered`] or [`FuturesUnordered`] with a unified
+/// interface, allowing the caller to choose between ordering and throughput
+/// at runtime without duplicating the polling loop.
+pub enum InFlightQueue<Fut: Future> {
+    Ordered(FuturesOrdered<Fut>),
+    Unordered(FuturesUnordered<Fut>),
+}
+
+impl<Fut: Future> InFlightQueue<Fut> {
+    pub fn new(preserve_ordering: bool) -> Self {
+        if preserve_ordering {
+            Self::Ordered(FuturesOrdered::new())
+        } else {
+            Self::Unordered(FuturesUnordered::new())
+        }
+    }
+
+    pub fn push(&mut self, fut: Fut) {
+        match self {
+            Self::Ordered(q) => q.push_back(fut),
+            Self::Unordered(q) => q.push(fut),
+        }
+    }
+
+    pub fn len(&self) -> usize {
+        match self {
+            Self::Ordered(q) => q.len(),
+            Self::Unordered(q) => q.len(),
+        }
+    }
+
+    pub fn is_empty(&self) -> bool {
+        match self {
+            Self::Ordered(q) => q.is_empty(),
+            Self::Unordered(q) => q.is_empty(),
+        }
+    }
+}
+
+impl<Fut: Future> Stream for InFlightQueue<Fut> {
+    type Item = Fut::Output;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        match self.get_mut() {
+            Self::Ordered(q) => Pin::new(q).poll_next(cx),
+            Self::Unordered(q) => Pin::new(q).poll_next(cx),
+        }
+    }
+}

src/topology/mod.rs

@@ -12,6 +12,7 @@ pub mod schema;
 
 pub mod builder;
 mod controller;
+mod in_flight_queue;
 mod ready_arrays;
 mod running;
 mod task;