diff --git a/Cargo.lock b/Cargo.lock index d172886ccc2..8f8059a5e0e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3976,7 +3976,7 @@ dependencies = [ [[package]] name = "dbsp" -version = "0.319.0" +version = "0.320.0" dependencies = [ "anyhow", "arc-swap", @@ -4066,7 +4066,7 @@ dependencies = [ [[package]] name = "dbsp_adapters" -version = "0.319.0" +version = "0.320.0" dependencies = [ "actix", "actix-codec", @@ -4210,7 +4210,7 @@ dependencies = [ [[package]] name = "dbsp_nexmark" -version = "0.319.0" +version = "0.320.0" dependencies = [ "anyhow", "ascii_table", @@ -5194,7 +5194,7 @@ dependencies = [ [[package]] name = "fda" -version = "0.319.0" +version = "0.320.0" dependencies = [ "anyhow", "arrow", @@ -5246,7 +5246,7 @@ dependencies = [ [[package]] name = "feldera-adapterlib" -version = "0.319.0" +version = "0.320.0" dependencies = [ "actix-web", "anyhow", @@ -5280,7 +5280,7 @@ dependencies = [ [[package]] name = "feldera-buffer-cache" -version = "0.319.0" +version = "0.320.0" dependencies = [ "crossbeam-utils", "enum-map", @@ -5308,7 +5308,7 @@ dependencies = [ [[package]] name = "feldera-datagen" -version = "0.319.0" +version = "0.320.0" dependencies = [ "anyhow", "async-channel 2.5.0", @@ -5334,7 +5334,7 @@ dependencies = [ [[package]] name = "feldera-fxp" -version = "0.319.0" +version = "0.320.0" dependencies = [ "bytecheck", "dbsp", @@ -5354,7 +5354,7 @@ dependencies = [ [[package]] name = "feldera-iceberg" -version = "0.319.0" +version = "0.320.0" dependencies = [ "anyhow", "chrono", @@ -5377,7 +5377,7 @@ dependencies = [ [[package]] name = "feldera-ir" -version = "0.319.0" +version = "0.320.0" dependencies = [ "proptest", "proptest-derive", @@ -5389,7 +5389,7 @@ dependencies = [ [[package]] name = "feldera-macros" -version = "0.319.0" +version = "0.320.0" dependencies = [ "prettyplease", "proc-macro2", @@ -5399,7 +5399,7 @@ dependencies = [ [[package]] name = "feldera-observability" -version = "0.319.0" +version = "0.320.0" dependencies = [ "actix-http", "awc", @@ -5414,7 +5414,7 @@ dependencies = [ [[package]] name = "feldera-rest-api" -version = "0.319.0" +version = "0.320.0" dependencies = [ "chrono", "feldera-observability", @@ -5433,7 +5433,7 @@ dependencies = [ [[package]] name = "feldera-samply" -version = "0.319.0" +version = "0.320.0" dependencies = [ "crossbeam", "feldera-size-of", @@ -5469,7 +5469,7 @@ dependencies = [ [[package]] name = "feldera-sqllib" -version = "0.319.0" +version = "0.320.0" dependencies = [ "arcstr", "base58", @@ -5513,7 +5513,7 @@ dependencies = [ [[package]] name = "feldera-storage" -version = "0.319.0" +version = "0.320.0" dependencies = [ "anyhow", "crossbeam", @@ -5536,7 +5536,7 @@ dependencies = [ [[package]] name = "feldera-types" -version = "0.319.0" +version = "0.320.0" dependencies = [ "actix-web", "anyhow", @@ -8813,7 +8813,7 @@ checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" [[package]] name = "pipeline-manager" -version = "0.319.0" +version = "0.320.0" dependencies = [ "actix-cors", "actix-files", @@ -10023,7 +10023,7 @@ dependencies = [ [[package]] name = "readers" -version = "0.319.0" +version = "0.320.0" dependencies = [ "async-std", "csv", @@ -11722,7 +11722,7 @@ checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" [[package]] name = "sltsqlvalue" -version = "0.319.0" +version = "0.320.0" dependencies = [ "dbsp", "feldera-sqllib", @@ -12193,7 +12193,7 @@ checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f" [[package]] name = "storage-test-compat" -version = "0.319.0" +version = "0.320.0" dependencies = [ "dbsp", "derive_more 1.0.0", diff --git a/Cargo.toml b/Cargo.toml index 3b433eb54a7..2985273bea9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [workspace.package] authors = ["Feldera Team "] -version = "0.319.0" +version = "0.320.0" license = "MIT OR Apache-2.0" homepage = "https://github.com/feldera/feldera" repository = "https://github.com/feldera/feldera" @@ -108,7 +108,7 @@ csv = "1.2.2" csv-core = "0.1.10" dashmap = "6.1.0" datafusion = "53.1" -dbsp = { path = "crates/dbsp", version = "0.319.0" } +dbsp = { path = "crates/dbsp", version = "0.320.0" } dbsp_nexmark = { path = "crates/nexmark" } deadpool-postgres = "0.14.1" # Feldera fork of delta-rs: upstream tag `rust-v0.32.3` + 2 patches, on branch @@ -133,20 +133,20 @@ erased-serde = "0.3.31" fake = "2.10" fastbloom = "0.14.0" fdlimit = "0.3.0" -feldera-buffer-cache = { version = "0.319.0", path = "crates/buffer-cache" } +feldera-buffer-cache = { version = "0.320.0", path = "crates/buffer-cache" } feldera-cloud1-client = "0.1.2" feldera-datagen = { path = "crates/datagen" } -feldera-fxp = { version = "0.319.0", path = "crates/fxp", features = ["dbsp"] } +feldera-fxp = { version = "0.320.0", path = "crates/fxp", features = ["dbsp"] } feldera-iceberg = { path = "crates/iceberg" } -feldera-observability = { version = "0.319.0", path = "crates/feldera-observability" } -feldera-macros = { version = "0.319.0", path = "crates/feldera-macros" } -feldera-sqllib = { version = "0.319.0", path = "crates/sqllib" } -feldera-storage = { version = "0.319.0", path = "crates/storage" } -feldera-types = { version = "0.319.0", path = "crates/feldera-types" } -feldera-rest-api = { version = "0.319.0", path = "crates/rest-api" } -feldera-ir = { version = "0.319.0", path = "crates/ir" } -feldera-adapterlib = { version = "0.319.0", path = "crates/adapterlib" } -feldera-samply = { version = "0.319.0", path = "crates/samply" } +feldera-observability = { version = "0.320.0", path = "crates/feldera-observability" } +feldera-macros = { version = "0.320.0", path = "crates/feldera-macros" } +feldera-sqllib = { version = "0.320.0", path = "crates/sqllib" } +feldera-storage = { version = "0.320.0", path = "crates/storage" } +feldera-types = { version = "0.320.0", path = "crates/feldera-types" } +feldera-rest-api = { version = "0.320.0", path = "crates/rest-api" } +feldera-ir = { version = "0.320.0", path = "crates/ir" } +feldera-adapterlib = { version = "0.320.0", path = "crates/adapterlib" } +feldera-samply = { version = "0.320.0", path = "crates/samply" } flate2 = "1.1.0" form_urlencoded = "1.2.0" fs_extra = "1.3.0" diff --git a/crates/adapters/src/controller.rs b/crates/adapters/src/controller.rs index 298f14760b1..e0a99ec77f5 100644 --- a/crates/adapters/src/controller.rs +++ b/crates/adapters/src/controller.rs @@ -3565,9 +3565,18 @@ impl CircuitThread { CheckpointRequest::Scheduled => (), CheckpointRequest::CheckpointCommand(callback) => callback(result.clone()), CheckpointRequest::SuspendCommand(callback) => { - self.controller.status.set_state(PipelineState::Terminated); - if let Err(e) = &result { - self.controller.error(e.clone(), None); + // Terminate the circuit only on a *successful* suspend. A + // failed suspend leaves the circuit intact and running; the + // `/suspend` handler reports it as `PipelinePhase::Failed` + // and stops the pipeline. Not marking `Terminated` here is + // what keeps `/status` from masking the failure as a clean + // `Suspended` during teardown: `get_status` reads the live + // controller state before the phase, so a terminated + // controller with desired status `Suspended` always reads + // back as `Suspended` (see `terminated_status`). + match &result { + Ok(_) => self.controller.status.set_state(PipelineState::Terminated), + Err(e) => self.controller.error(e.clone(), None), } callback(result.clone().map(|_| ())) } diff --git a/crates/adapters/src/controller/test.rs b/crates/adapters/src/controller/test.rs index 6f2e34368a9..f19f98f3cde 100644 --- a/crates/adapters/src/controller/test.rs +++ b/crates/adapters/src/controller/test.rs @@ -431,9 +431,40 @@ fn collect_endpoint_records(controller: &Controller, n: usize) -> Vec { .collect() } +/// Default budget for [`wait_for_records`] to observe the expected output. +const OUTPUT_TIMEOUT_MS: u128 = 10_000; + +/// Longer output budget for waits taken while a suspend is in progress. +/// +/// While a suspend is pending, the controller advances only the barrier inputs +/// while re-attempting the checkpoint each step, so under CI CPU contention +/// output can lag far behind input even though stepping keeps making forward +/// progress. Using the default [`OUTPUT_TIMEOUT_MS`] in that window makes the +/// suspend tests flake. Matches the 100s budget already used for the +/// suspend-completion `recv_timeout` in these tests. +const SUSPEND_OUTPUT_TIMEOUT_MS: u128 = 100_000; + +/// Like `println!`, but prefixes a UTC timestamp so test output interleaves +/// legibly with the controller's tracing logs when diagnosing timing issues. +macro_rules! tprintln { + ($($arg:tt)*) => { + println!("[{}] {}", chrono::Utc::now().format("%H:%M:%S%.6f"), format_args!($($arg)*)) + }; +} + +/// Wait until every output endpoint has received at least the expected number +/// of records, then verify the counts match exactly, using the default +/// [`OUTPUT_TIMEOUT_MS`] budget. #[track_caller] fn wait_for_records(controller: &Controller, expect_n: &[usize]) { - println!("waiting for {expect_n:?} records..."); + wait_for_records_within(controller, expect_n, OUTPUT_TIMEOUT_MS); +} + +/// Like [`wait_for_records`], but waits up to `timeout_ms` for the records to +/// arrive. +#[track_caller] +fn wait_for_records_within(controller: &Controller, expect_n: &[usize], timeout_ms: u128) { + tprintln!("waiting for {expect_n:?} records..."); let n = expect_n.len(); let mut last_n = repeat_n(0, n).collect::>(); wait( @@ -441,7 +472,7 @@ fn wait_for_records(controller: &Controller, expect_n: &[usize]) { let new_n = collect_endpoint_records(controller, n); for i in 0..n { if new_n[i] > last_n[i] { - println!("received {} records on test_output{}", new_n[i], i + 1); + tprintln!("received {} records on test_output{}", new_n[i], i + 1); } } last_n = new_n; @@ -450,7 +481,7 @@ fn wait_for_records(controller: &Controller, expect_n: &[usize]) { .zip(expect_n.iter()) .all(|(&last, &expect)| last >= expect) }, - 10_000, + timeout_ms, ) .unwrap(); @@ -2028,25 +2059,25 @@ fn suspend_barrier() { .from_writer(&input_file); // Write records to the input file. - println!("Writing records 0..4000"); + tprintln!("Writing records 0..4000"); for id in 0..4000 { writer.serialize(TestStruct::for_id(id as u32)).unwrap(); } writer.flush().unwrap(); // Start pipeline. - println!("start pipeline"); + tprintln!("start pipeline"); let controller = start_controller(&storage_dir, &[5000]); // Wait for the records that are not in the checkpoint to be // processed or replayed. - println!("wait for 4000 records 0..4000"); + tprintln!("wait for 4000 records 0..4000"); wait_for_records(&controller, &[4000]); // Suspend. let (sender, receiver) = mpsc::channel(); - println!("start suspend"); + tprintln!("start suspend"); let suspend_request_step = controller.status().global_metrics.total_initiated_steps(); controller.start_suspend(Box::new(move |result| sender.send(result).unwrap())); @@ -2068,7 +2099,7 @@ fn suspend_barrier() { assert_bounded_suspend_steps(&controller, suspend_request_step); // Stop controller. - println!("stop controller"); + tprintln!("stop controller"); controller.stop().unwrap(); // Read output and compare. Our output adapter, which is not @@ -2082,7 +2113,7 @@ fn suspend_barrier() { let controller = start_controller(&storage_dir, &[5000]); wait_for_records(&controller, &[5000]); - println!("start suspend"); + tprintln!("start suspend"); let (sender, receiver) = mpsc::channel(); let mut sender = Some(sender); @@ -2092,13 +2123,16 @@ fn suspend_barrier() { let start = (i + 5) * 1000; let end = start + 1000; - println!("writing records {start}..{end}"); + tprintln!("writing records {start}..{end}"); for id in start..end { writer.serialize(TestStruct::for_id(id as u32)).unwrap(); } writer.flush().unwrap(); - println!("waiting for {end} records"); - wait_for_records(&controller, &[end]); + tprintln!("waiting for {end} records"); + // After the first iteration a suspend is pending: the controller only + // advances barrier inputs while re-attempting the checkpoint each step, + // so output can lag under CI CPU load. Use the longer budget. + wait_for_records_within(&controller, &[end], SUSPEND_OUTPUT_TIMEOUT_MS); check_file_contents(&(output_path(&storage_dir, 0)), 5000..end); if let Some(sender) = sender.take() { @@ -2157,7 +2191,7 @@ fn suspend_multiple_barriers(n_inputs: usize) { .collect::>(); // Write records to the input files. - println!("Writing 1000 records to each of {n_inputs} files"); + tprintln!("Writing 1000 records to each of {n_inputs} files"); for writer in writers.iter_mut().take(n_inputs) { for id in 0..1000 { writer.serialize(TestStruct::for_id(id as u32)).unwrap(); @@ -2166,7 +2200,7 @@ fn suspend_multiple_barriers(n_inputs: usize) { } // Start pipeline. - println!("start pipeline"); + tprintln!("start pipeline"); // The barrier for input 0 is record 0, // for input 1 is record 1000, @@ -2177,7 +2211,7 @@ fn suspend_multiple_barriers(n_inputs: usize) { // Wait for the first 1000 records in each file to be read and copied to the // output. - println!("wait for 1000 records in each file"); + tprintln!("wait for 1000 records in each file"); let mut written = repeat_n(1000, n_inputs).collect::>(); wait_for_records(&controller, &written); @@ -2187,7 +2221,7 @@ fn suspend_multiple_barriers(n_inputs: usize) { // we're past all the barriers; otherwise, it will not complete due to // barriers, since each input only has 1000 records so far. let (sender, receiver) = mpsc::channel(); - println!("start suspend"); + tprintln!("start suspend"); let suspend_request_step = controller.status().global_metrics.total_initiated_steps(); controller.start_suspend(Box::new(move |result| sender.send(result).unwrap())); @@ -2212,7 +2246,7 @@ fn suspend_multiple_barriers(n_inputs: usize) { receiver.try_recv().unwrap_err(); // Write 1000 more records to the `next` input. - println!("writing 1000 more records to test_input{}", next + 1); + tprintln!("writing 1000 more records to test_input{}", next + 1); for id in written[next]..written[next] + 1000 { writers[next] .serialize(TestStruct::for_id(id as u32)) @@ -2228,9 +2262,12 @@ fn suspend_multiple_barriers(n_inputs: usize) { // We won't get any more records on output from inputs that have reached // their barrier, so the writes to inputs 0 and 1 won't have any effect // here. - println!("total written: {written:?}"); + tprintln!("total written: {written:?}"); let expect = expectations(&written, &barriers); - wait_for_records(&controller, &expect); + // A suspend is pending here: the controller only advances barrier inputs + // while re-attempting the checkpoint each step, so output can lag far + // behind under CI CPU load. Use the longer budget. + wait_for_records_within(&controller, &expect, SUSPEND_OUTPUT_TIMEOUT_MS); for (i, expectation) in expect.iter().enumerate().take(n_inputs) { check_file_contents(&output_path(&storage_dir, i), 0..*expectation); } @@ -2244,11 +2281,11 @@ fn suspend_multiple_barriers(n_inputs: usize) { assert_bounded_suspend_steps(&controller, suspend_request_step); // Stop controller. - println!("stop controller"); + tprintln!("stop controller"); controller.stop().unwrap(); // Check output one more time. - println!("check output one more time now that controller is stopped"); + tprintln!("check output one more time now that controller is stopped"); let expect = expectations(&written, &barriers); for (i, e) in expect.iter().enumerate().take(n_inputs) { check_file_contents(&output_path(&storage_dir, i), 0..*e); @@ -2256,7 +2293,7 @@ fn suspend_multiple_barriers(n_inputs: usize) { // Now restart the controller and wait for all the records that we wrote // beyond the barriers get copied to output (and nothing else). - println!("restart controller and wait for records beyond the barriers"); + tprintln!("restart controller and wait for records beyond the barriers"); let controller = start_controller(&storage_dir, &barriers); wait_for_records(&controller, &written); for i in 0..n_inputs { diff --git a/crates/adapters/src/server.rs b/crates/adapters/src/server.rs index 868b5431221..e6f821fcfa5 100644 --- a/crates/adapters/src/server.rs +++ b/crates/adapters/src/server.rs @@ -1419,6 +1419,47 @@ async fn status_handler( get_status(&state) } +#[allow(clippy::result_large_err)] +/// Reports the runtime status of a controller whose circuit has reached +/// [`PipelineState::Terminated`]. +/// +/// A successful suspend terminates the circuit before the server installs +/// [`PipelinePhase::Suspended`] and deallocates the controller (see `/suspend`), +/// so a status poll landing in that window observes the still-registered +/// controller in `Terminated`. When a suspend was requested the pipeline is +/// converging to `Suspended`, so report that clean status rather than a +/// spurious `PipelineTerminated` error that the pipeline manager would record +/// as a failed execution. +/// +/// This branch is reached only for a *successful* suspend: a failed suspend +/// deliberately leaves the circuit running (see the `SuspendCommand` handler in +/// `controller.rs`) so that it never masquerades here as a clean `Suspended`; +/// the `/suspend` handler reports it as [`PipelinePhase::Failed`] instead. Any +/// termination without a suspend request is an unexpected, fatal termination +/// and stays an error. +fn terminated_status( + runtime_desired_status: RuntimeDesiredStatus, + storage_status_details: Option, +) -> Result { + if matches!(runtime_desired_status, RuntimeDesiredStatus::Suspended) { + Ok(ExtendedRuntimeStatus { + runtime_status: RuntimeStatus::Suspended, + runtime_status_details: json!(""), + runtime_desired_status, + storage_status_details, + }) + } else { + Err(ExtendedRuntimeStatusError { + status_code: StatusCode::INTERNAL_SERVER_ERROR, + error: feldera_types::error::ErrorResponse { + message: "Pipeline has been terminated.".to_string(), + error_code: Cow::from("PipelineTerminated"), + details: json!({}), + }, + }) + } +} + #[allow(clippy::result_large_err)] fn get_status(state: &ServerState) -> Result { // Runtime desired status @@ -1476,14 +1517,9 @@ fn get_status(state: &ServerState) -> Result Err(ExtendedRuntimeStatusError { - status_code: StatusCode::INTERNAL_SERVER_ERROR, - error: feldera_types::error::ErrorResponse { - message: "Pipeline has been terminated.".to_string(), - error_code: Cow::from("PipelineTerminated"), - details: json!({}), - }, - }), + PipelineState::Terminated => { + terminated_status(runtime_desired_status, storage_status_details) + } }; } Err(_) => { @@ -2135,10 +2171,12 @@ async fn suspend(state: WebData) -> Result) { + let mut suspend_error = None; loop { if let Ok(controller) = state.controller() { if let Err(error) = controller.async_suspend().await { error!("controller suspend failed ({error})"); + suspend_error = Some(error); } break; } @@ -2151,7 +2189,17 @@ async fn suspend(state: WebData) -> Result PipelinePhase::Failed(error), + None => PipelinePhase::Suspended, + }); if let Ok(controller) = state.take_controller() && let Err(error) = controller.async_stop().await { @@ -3146,7 +3194,7 @@ impl StoredStatus { /// off. #[cfg(test)] mod test_http_helpers { - use super::{ServerArgs, ServerState, bootstrap, build_app, parse_config}; + use super::{ServerArgs, ServerState, bootstrap, build_app, parse_config, terminated_status}; use crate::{ controller::ControllerBuilder, server::{InitializationState, PipelinePhase}, @@ -3181,6 +3229,26 @@ mod test_http_helpers { use tempfile::NamedTempFile; use uuid::Uuid; + /// A successful suspend terminates the circuit before the server reports + /// `PipelinePhase::Suspended`, so `terminated_status` must report a clean + /// `Suspended` while a suspend is in progress instead of the spurious + /// `PipelineTerminated` error that flaked `suspend_and_resume_demos`. + /// + /// A *failed* suspend never reaches `terminated_status`: the `SuspendCommand` + /// handler leaves the circuit running instead of terminating it, and the + /// `/suspend` handler reports the failure as `PipelinePhase::Failed`. + #[test] + fn terminated_status_reports_suspended_while_suspending() { + let status = terminated_status(RuntimeDesiredStatus::Suspended, None) + .expect("a suspending pipeline must not surface PipelineTerminated"); + assert!(matches!(status.runtime_status, RuntimeStatus::Suspended)); + + // An unexpected termination (no suspend requested) stays a fatal error. + let error = terminated_status(RuntimeDesiredStatus::Running, None) + .expect_err("an unexpected termination must remain PipelineTerminated"); + assert_eq!(error.error.error_code.as_ref(), "PipelineTerminated"); + } + pub(super) async fn get_stats(server: &TestServer) -> ExternalControllerStatus { server .get("/stats") diff --git a/crates/dbsp/src/algebra/zset.rs b/crates/dbsp/src/algebra/zset.rs index b4c1b11db22..21a93fee1d4 100644 --- a/crates/dbsp/src/algebra/zset.rs +++ b/crates/dbsp/src/algebra/zset.rs @@ -250,6 +250,11 @@ pub trait ZSet: IndexedZSet { /// 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 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 index fdc1a7937a6..91f59bccf5a 100644 --- 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 index 28a7f416e7f..ae72ce47ef2 100644 --- 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::(); + + self.inner().dyn_positive_mono(&factories).typed() + } + #[track_caller] pub fn hash_distinct(&self) -> Self { let factories = HashDistinctFactories::new::(); @@ -1244,6 +1251,13 @@ where self.inner().dyn_distinct_mono(&factories).typed() } + #[track_caller] + pub fn positive(&self) -> Self { + let factories = DistinctFactories::new::(); + + self.inner().dyn_positive_mono(&factories).typed() + } + #[track_caller] pub fn hash_distinct(&self) -> Self { let factories = HashDistinctFactories::new::(); diff --git a/crates/dbsp/src/operator/distinct.rs b/crates/dbsp/src/operator/distinct.rs index 320429d9142..e5a7a1cab10 100644 --- 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 Stream -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 + where + Z: crate::typed_batch::ZSet, + { + let factories = + crate::operator::dynamic::distinct::DistinctFactories::new::(); + + self.inner().dyn_positive(&factories).typed() + } } diff --git a/crates/dbsp/src/operator/dynamic/distinct.rs b/crates/dbsp/src/operator/dynamic/distinct.rs index 3187b11ee89..0993728c83b 100644 --- a/crates/dbsp/src/operator/dynamic/distinct.rs +++ b/crates/dbsp/src/operator/dynamic/distinct.rs @@ -14,7 +14,7 @@ use crate::{ DBData, Runtime, Timestamp, ZWeight, algebra::{ AddByRef, HasOne, HasZero, IndexedZSet, Lattice, OrdIndexedZSet, OrdIndexedZSetFactories, - PartialOrder, ZRingValue, + PartialOrder, }, circuit::{ Circuit, Scope, Stream, WithClock, @@ -49,6 +49,45 @@ use super::{MonoIndexedZSet, MonoZSet}; circuit_cache_key!(DistinctId(StreamId => Stream)); circuit_cache_key!(DistinctIncrementalId(StreamId => Stream)); +circuit_cache_key!(PositiveIncrementalId(StreamId => Stream)); + +/// Abstract semantics of the generic `distinct` operator. +pub trait DistinctSemantics: 'static { + /// Name of the incremental operator specialized for the root scope. + const ROOT_SCOPE_NAME: &'static str; + /// Name of the incremental operator for nested scopes. + const NESTED_SCOPE_NAME: &'static str; + + /// Maps the weight `w` of a tuple in the input to the weight of + /// the same tuple in the output. + fn output_weight(w: ZWeight) -> ZWeight; +} + +/// [`DistinctSemantics`] of the `distinct` operator: tuples with positive +/// weight get weight 1. +pub struct ZeroOrOneWeight; + +impl DistinctSemantics for ZeroOrOneWeight { + const ROOT_SCOPE_NAME: &'static str = "DistinctIncrementalTotal"; + const NESTED_SCOPE_NAME: &'static str = "DistinctIncremental"; + + fn output_weight(w: ZWeight) -> ZWeight { + if w > 0 { 1 } else { 0 } + } +} + +/// [`DistinctSemantics`] of the `positive` operator: tuples with positive +/// weight keep their weight. +pub struct PositiveWeight; + +impl DistinctSemantics for PositiveWeight { + const ROOT_SCOPE_NAME: &'static str = "PositiveIncrementalTotal"; + const NESTED_SCOPE_NAME: &'static str = "PositiveIncremental"; + + fn output_weight(w: ZWeight) -> ZWeight { + w.max(0) + } +} pub struct DistinctFactories { pub input_factories: Z::Factories, @@ -189,6 +228,13 @@ impl Stream { self.dyn_distinct(factories) } + pub fn dyn_positive_mono( + &self, + factories: &DistinctFactories, + ) -> Stream { + self.dyn_positive(factories) + } + pub fn dyn_hash_distinct_mono( &self, factories: &HashDistinctFactories, @@ -221,6 +267,13 @@ impl Stream { self.dyn_distinct(factories) } + pub fn dyn_positive_mono( + &self, + factories: &DistinctFactories::Time>, + ) -> Stream { + self.dyn_positive(factories) + } + pub fn dyn_hash_distinct_mono( &self, factories: &HashDistinctFactories::Time>, @@ -307,9 +360,42 @@ where }) } + /// See [`Stream::positive`]. + pub fn dyn_positive(&self, factories: &DistinctFactories) -> Stream + where + Z: IndexedZSet + Send, + { + let circuit = self.circuit(); + circuit.region("positive", || { + let stream = self.try_sharded_version(); + + circuit + .cache_get_or_insert_with(PositiveIncrementalId::new(stream.stream_id()), || { + stream.dyn_positive_inner(factories) + }) + .clone() + }) + } + pub fn dyn_distinct_inner(&self, factories: &DistinctFactories) -> Stream where Z: IndexedZSet + Send, + { + self.distinct_inner_generic::(factories) + .mark_distinct() + } + + pub fn dyn_positive_inner(&self, factories: &DistinctFactories) -> Stream + where + Z: IndexedZSet + Send, + { + self.distinct_inner_generic::(factories) + } + + fn distinct_inner_generic(&self, factories: &DistinctFactories) -> Stream + where + Z: IndexedZSet + Send, + S: DistinctSemantics, { let circuit = self.circuit(); @@ -318,7 +404,7 @@ where if circuit.root_scope() == 0 { // Use an implementation optimized to work in the root scope. circuit.add_binary_operator( - StreamingBinaryWrapper::new(DistinctIncrementalTotal::new( + StreamingBinaryWrapper::new(DistinctIncrementalTotal::<_, _, S>::new( Location::caller(), &factories.input_factories, )), @@ -339,7 +425,7 @@ where // └───────────┘ └─────┘ └─────┘ └───────────────────┘ // ``` circuit.add_binary_operator( - StreamingBinaryWrapper::new(DistinctIncremental::new( + StreamingBinaryWrapper::new(DistinctIncremental::<_, _, _, S>::new( Location::caller(), &factories.input_factories, &factories.aux_factories, @@ -356,7 +442,6 @@ where ) } .mark_sharded() - .mark_distinct() } } @@ -403,14 +488,15 @@ where } } -/// Incremental version of the distinct operator that only works in the -/// top-level scope (i.e., for totally ordered timestamps). +/// Incremental version of the distinct and positive operators that only works +/// in the top-level scope (i.e., for totally ordered timestamps). /// /// Takes a stream `a` of changes to relation `A` and a stream with delayed /// value of `A`: `z^-1(A) = a.integrate().delay()` and computes -/// `distinct(A) - distinct(z^-1(A))` incrementally, by only considering -/// values in the support of `a`. -struct DistinctIncrementalTotal { +/// `f(A) - f(z^-1(A))` incrementally, by only considering values in the +/// support of `a`, where `f` applies [`DistinctSemantics::output_weight`] to +/// the weight of every tuple. +struct DistinctIncrementalTotal { input_factories: Z::Factories, location: &'static Location<'static>, @@ -423,10 +509,10 @@ struct DistinctIncrementalTotal { chunk_size: usize, first_chunk_size: usize, - _type: PhantomData<(Z, I)>, + _type: PhantomData<(Z, I, S)>, } -impl DistinctIncrementalTotal { +impl DistinctIncrementalTotal { pub fn new(location: &'static Location<'static>, input_factories: &Z::Factories) -> Self { Self { input_factories: input_factories.clone(), @@ -470,13 +556,14 @@ impl DistinctIncrementalTotal { } } -impl Operator for DistinctIncrementalTotal +impl Operator for DistinctIncrementalTotal where Z: IndexedZSet, I: 'static, + S: DistinctSemantics, { fn name(&self) -> Cow<'static, str> { - Cow::from("DistinctIncrementalTotal") + Cow::from(S::ROOT_SCOPE_NAME) } fn location(&self) -> OperatorLocation { @@ -495,10 +582,11 @@ where } } -impl StreamingBinaryOperator>, I, Z> for DistinctIncrementalTotal +impl StreamingBinaryOperator>, I, Z> for DistinctIncrementalTotal where Z: IndexedZSet, I: WithSnapshot + 'static, + S: DistinctSemantics, { fn eval( self: Rc, @@ -555,15 +643,13 @@ where let new_weight = old_weight.add_by_ref(&w); - if old_weight.le0() { - // Weight changes from non-positive to positive. - if new_weight.ge0() && !new_weight.is_zero() { - builder.push_val_diff(v, ZWeight::one().erase()); - any_values = true; - } - } else if new_weight.le0() { - // Weight changes from positive to non-positive. - builder.push_val_diff(v, ZWeight::one().neg().erase()); + // Emit the change in the tuple's output weight. For + // `distinct` it is nonzero only when the integrated + // weight changes sign; for `positive` also when a + // positive weight changes value. + let diff = S::output_weight(new_weight) - S::output_weight(old_weight); + if !diff.is_zero() { + builder.push_val_diff(v, diff.erase()); any_values = true; } @@ -574,12 +660,15 @@ where delta_cursor.step_val(); } } else { + // The key is missing from the integral: the weight + // if copied from the delta while delta_cursor.val_valid() { let new_weight = **delta_cursor.weight(); debug_assert!(!new_weight.is_zero()); - if new_weight.ge0() { - builder.push_val_diff(delta_cursor.val(), ZWeight::one().erase()); + let output_weight = S::output_weight(new_weight); + if !output_weight.is_zero() { + builder.push_val_diff(delta_cursor.val(), output_weight.erase()); any_values = true; } @@ -612,7 +701,7 @@ where type KeysOfInterest = BTreeMap, R>>>; #[derive(SizeOf)] -struct DistinctIncremental +struct DistinctIncremental where Z: IndexedZSet, T: WithSnapshot, @@ -649,15 +738,16 @@ where chunk_size: usize, first_chunk_size: usize, - _type: PhantomData<(Z, T)>, + _type: PhantomData<(Z, T, S)>, } -impl DistinctIncremental +impl DistinctIncremental where Z: IndexedZSet, T: WithSnapshot, T::Batch: ZBatchReader, Clk: WithClock