feldera · mihaibudiu · Jul 11, 2026 · Jul 10, 2026
diff --git a/crates/dbsp/src/algebra/zset.rs b/crates/dbsp/src/algebra/zset.rs
@@ -250,6 +250,11 @@ pub trait ZSet: IndexedZSet<Val = DynUnit> {
     /// negative, so the result can be zero even if the Z-set contains nonzero
     /// weights.
     fn weighted_count(&self, sum: &mut Self::R);
+
+    /// Returns a Z-set that contains all elements with positive weights from
+    /// `self` with their weights preserved.
+    #[cfg(test)]
+    fn positive(&self) -> Self;
 }
 
 impl<Z, K> ZSet for Z
@@ -266,6 +271,24 @@ where
             cursor.step_key();
         }
     }
+
+    #[cfg(test)]
+    fn positive(&self) -> Self {
+        let factories = self.factories();
+        let mut builder = Self::Builder::with_capacity(&factories, self.key_count(), self.len());
+        let mut cursor = self.cursor();
+
+        while cursor.key_valid() {
+            let weight = **cursor.weight();
+            if weight > 0 {
+                builder.push_val_diff(cursor.val(), weight.erase());
+                builder.push_key(cursor.key());
+            }
+            cursor.step_key();
+        }
+
+        builder.done()
+    }
 }
 
 #[cfg(test)]

diff --git a/crates/dbsp/src/circuit/circuit_builder.rs b/crates/dbsp/src/circuit/circuit_builder.rs
@@ -544,6 +544,10 @@ dyn_clone::clone_trait_object!(StreamMetadata);
 ///     data.  It sets each record's weight to 1 if it is positive and drops the
 ///     others.
 ///
+/// The "positive" operator on a Z-set keeps positive weights unchanged and
+/// maps all other weights to 0.  It differs from "distinct" only in that it
+/// does not clamp positive weights to 1.
+///
 /// ## Join on equal keys
 ///
 /// A DBSP equi-join takes batches `a` and `b` as input, finds all pairs of a

diff --git a/crates/dbsp/src/mono.rs b/crates/dbsp/src/mono.rs
@@ -762,6 +762,13 @@ where
         self.inner().dyn_distinct_mono(&factories).typed()
     }
 
+    #[track_caller]
+    pub fn positive(&self) -> Self {
+        let factories = DistinctFactories::new::<K, ()>();
+
+        self.inner().dyn_positive_mono(&factories).typed()
+    }
+
     #[track_caller]
     pub fn hash_distinct(&self) -> Self {
         let factories = HashDistinctFactories::new::<K, ()>();
@@ -1244,6 +1251,13 @@ where
         self.inner().dyn_distinct_mono(&factories).typed()
     }
 
+    #[track_caller]
+    pub fn positive(&self) -> Self {
+        let factories = DistinctFactories::new::<K, ()>();
+
+        self.inner().dyn_positive_mono(&factories).typed()
+    }
+
     #[track_caller]
     pub fn hash_distinct(&self) -> Self {
         let factories = HashDistinctFactories::new::<K, ()>();

diff --git a/crates/dbsp/src/operator/distinct.rs b/crates/dbsp/src/operator/distinct.rs
@@ -22,14 +22,7 @@ where
 
         self.inner().dyn_stream_distinct(&factories).typed()
     }
-}
 
-impl<C, Z> Stream<C, Z>
-where
-    C: Circuit,
-    Z: IndexedZSet,
-    Z::InnerBatch: Send,
-{
     /// Incrementally deduplicate input stream.
     ///
     /// This is an incremental version of the
@@ -62,4 +55,25 @@ where
 
         self.inner().dyn_has_distinct(&factories).typed()
     }
+
+    /// Incrementally filter out tuples with non-positive weights.
+    ///
+    /// Given a stream of changes to relation `A`, computes a stream of
+    /// changes to relation `A'` that contains every `(key, weight)` tuple of
+    /// `A` with `weight > 0`, with the weight unchanged.
+    ///
+    /// This operator is like [`distinct`](`Self::distinct`), except that it
+    /// preserves the weights of the retained tuples instead of clamping them
+    /// to 1.
+    #[cfg(not(feature = "backend-mode"))]
+    #[track_caller]
+    pub fn positive(&self) -> Stream<C, Z>
+    where
+        Z: crate::typed_batch::ZSet,
+    {
+        let factories =
+            crate::operator::dynamic::distinct::DistinctFactories::new::<Z::Key, Z::Val>();
+
+        self.inner().dyn_positive(&factories).typed()
+    }
 }