feldera
diff --git a/‎.gitattributes‎
Lines changed: 1 addition & 1 deletion b/‎.gitattributes‎
Lines changed: 1 addition & 1 deletion
diff --git a/‎Cargo.lock‎
Lines changed: 154 additions & 34 deletions b/‎Cargo.lock‎
Lines changed: 154 additions & 34 deletions
diff --git a/‎benches/ldbc-graphalytics/bfs.rs‎
Lines changed: 7 additions & 3 deletions b/‎benches/ldbc-graphalytics/bfs.rs‎
Lines changed: 7 additions & 3 deletions
diff --git a/‎benches/ldbc-graphalytics/main.rs‎
Lines changed: 1 addition & 1 deletion b/‎benches/ldbc-graphalytics/main.rs‎
Lines changed: 1 addition & 1 deletion
diff --git a/‎benches/ldbc-graphalytics/pagerank.rs‎
Lines changed: 8 additions & 8 deletions b/‎benches/ldbc-graphalytics/pagerank.rs‎
Lines changed: 8 additions & 8 deletions
diff --git a/‎src/operator/mod.rs‎
Lines changed: 1 addition & 3 deletions b/‎src/operator/mod.rs‎
Lines changed: 1 addition & 3 deletions
@@ -1,2 +1,2 @@
-Cargo.lock -diff linguist-generated=true
+Cargo.lock merge=binary linguist-generated=true
 proptest-regressions/**/* -diff linguist-generated=true
@@ -6,7 +6,11 @@
 //! [A.1]: https://arxiv.org/pdf/2011.15028v4.pdf#section.A.1
 
 use crate::data::{Distance, DistanceSet, Edges, Node, Vertices};
-use dbsp::{operator::recursive::RecursiveStreams, time::NestedTimestamp32, Circuit, Stream};
+use dbsp::{
+    operator::{recursive::RecursiveStreams, FilterMap},
+    time::NestedTimestamp32,
+    Circuit, Stream,
+};
 
 type Distances<P> = Stream<Circuit<P>, DistanceSet>;
 
@@ -48,10 +52,10 @@ where
         .expect("failed to build dfs recursive scope");
 
     // Collect all reachable nodes
-    let reachable_nodes = distances.map_keys(|&(node, _)| node);
+    let reachable_nodes = distances.map(|&(node, _)| node);
     // Find all unreachable nodes (vertices not included in `distances`) and give them a weight of -1
     let unreachable_nodes =
-        antijoin(&vertices, &reachable_nodes).map_keys(|&node| (node, i64::MAX as Distance));
+        antijoin(&vertices, &reachable_nodes).map(|&node| (node, i64::MAX as Distance));
 
     distances.plus(&unreachable_nodes)
 }
 
@@ -61,7 +61,7 @@ fn main() {
         "using dataset {} with {} vertices and {} edges",
         dataset.name, properties.vertices, properties.edges,
     );
-    Runtime::run(1, move |_runtime, _index| {
+    Runtime::run(1, move || {
         print!("building dataflow... ");
         io::stdout().flush().unwrap();
         let start = Instant::now();
 
@@ -2,7 +2,7 @@ use crate::data::{Edges, Node, Rank, RankPairs, RankSet, Ranks, Streamed, Vertic
 use dbsp::{
     algebra::HasOne,
     circuit::operator_traits::Data,
-    operator::{DelayedFeedback, Generator},
+    operator::{DelayedFeedback, FilterMap, Generator},
     trace::{
         layers::Trie, ord::indexed_zset_batch::OrdIndexedZSetBuilder, Batch, BatchReader, Batcher,
         Builder, Cursor,
@@ -80,10 +80,10 @@ where
     });
 
     // Calculate the teleport, `(1 - 𝑑) ÷ |V|`
-    let teleport =
-        total_vertices.map_values::<OrdIndexedZSet<_, _, _>, _>(move |&(), &total_vertices| {
-            F64::new((1.0 - damping_factor) / total_vertices as f64)
-        });
+    let teleport = total_vertices.map_index(move |(&node, &total_vertices)| {
+        let teleport = F64::new((1.0 - damping_factor) / total_vertices as f64);
+        (node, teleport)
+    });
 
     // Count the number of outgoing edges for each node
     let outgoing_sans_dangling = edges.apply(|weights| {
@@ -112,7 +112,7 @@ where
         vertices.minus(&outgoing_sans_dangling.semijoin_stream_core(&vertices, |&node, _| node));
 
     // Make the edge weights for all dangling nodes, we only need to calculate this once
-    let dangling_edge_weights = dangling_nodes.map_keys(|&node| (node, F64::new(0.0)));
+    let dangling_edge_weights = dangling_nodes.map(|&node| (node, F64::new(0.0)));
 
     let reversed = if directed {
         edges.apply(|edges| {
@@ -134,14 +134,14 @@ where
 
     // Undirected graphs already have all the edges we need
     } else {
-        edges.map_values(|_, _| ())
+        edges.map(|(&src, _)| src)
     }
     .distinct();
 
     // Find vertices without any incoming edges
     let zero_incoming = vertices
         .minus(&reversed.semijoin::<(), _, _, _>(&vertices, |&node, _| node))
-        .map_keys::<OrdZSet<_, _>, _>(|&node| ((), (node, F64::new(0.0))));
+        .map(|&node| ((), (node, F64::new(0.0))));
 
     // Create a stream containing one `((), 0)` pair
     let zero = vertices
 
@@ -37,13 +37,11 @@ pub use apply::Apply;
 pub use condition::Condition;
 pub use delta0::Delta0;
 pub use distinct::Distinct;
-pub use filter::FilterKeys;
-pub use filter_map::FilterMapKeys;
+pub use filter_map::{FilterKeys, FilterMap, FilterVals, FlatMap, Map, MapKeys};
 pub use generator::{Generator, GeneratorNested};
 pub use index::Index;
 pub use inspect::Inspect;
 pub use join::Join;
-pub use map::{MapKeys, MapValues};
 pub use neg::UnaryMinus;
 pub use plus::{Minus, Plus};
 pub use sum::Sum;
Original file line number	Diff line number	Diff line change
`@@ -1,2 +1,2 @@`
`1`		`-Cargo.lock -diff linguist-generated=true`
	`1`	`+Cargo.lock merge=binary linguist-generated=true`
`2`	`2`	`proptest-regressions/*/ -diff linguist-generated=true`