feldera · vpopescu · Jul 17, 2026 · Jul 17, 2026 · Jul 17, 2026
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/crates/dbsp/Cargo.toml b/crates/dbsp/Cargo.toml
@@ -99,10 +99,15 @@ feldera-buffer-cache = { workspace = true }
 memory-stats = { workspace = true }
 serde_json_path_to_error = { workspace = true }
 flate2 = { workspace = true }
-nix = { workspace = true, features = ["process", "time"] }
 thread-id = { workspace = true }
 pin-project-lite = { workspace = true }
 
+[target.'cfg(unix)'.dependencies]
+nix = { workspace = true, features = ["process", "time"] }
+
+[target.'cfg(windows)'.dependencies]
+windows-sys = { version = "0.61", features = ["Win32_Foundation", "Win32_Storage_FileSystem", "Win32_System_IO", "Win32_System_Threading"] }
+
 [dev-dependencies]
 rand = { workspace = true }
 rand_distr = { workspace = true }
@@ -115,14 +120,16 @@ csv = { workspace = true }
 tar = { workspace = true }
 zstd = { workspace = true }
 criterion = { workspace = true }
-pprof = { workspace = true, features = ["flamegraph", "criterion"] }
 rand_xoshiro = { workspace = true }
 indicatif = { workspace = true }
 reqwest = { workspace = true, features = ["blocking"] }
 chrono = { workspace = true, features = ["serde"] }
 tracing-subscriber = { version = "0.3.20", features = ["env-filter"] }
 tarpc = { workspace = true, features = ["full"] }
 
+[target.'cfg(unix)'.dev-dependencies]
+pprof = { workspace = true, features = ["flamegraph", "criterion"] }
+
 [[bench]]
 name = "galen"
 harness = false

diff --git a/crates/dbsp/benches/column_layer.rs b/crates/dbsp/benches/column_layer.rs
@@ -10,6 +10,7 @@ use dbsp::{
     },
     utils::consolidate,
 };
+#[cfg(unix)]
 use pprof::criterion::{Output, PProfProfiler};
 use rand::{Rng, SeedableRng, distributions::Standard, prelude::Distribution};
 use rand_xoshiro::Xoshiro256StarStar;
@@ -233,9 +234,18 @@ leaf_benches! {
     "100,000,000-erased" = [Leaf]100_000_000,
 }
 
+#[cfg(unix)]
 criterion_group!(
     name = benches;
     config = Criterion::default().with_profiler(PProfProfiler::new(300, Output::Flamegraph(None)));
     targets = column_leaf, merge_ordered_column_leaf_builder
 );
+
+#[cfg(not(unix))]
+criterion_group!(
+    name = benches;
+    config = Criterion::default();
+    targets = column_leaf, merge_ordered_column_leaf_builder
+);
+
 criterion_main!(benches);
diff --git a/crates/dbsp/src/circuit/circuit_builder.rs b/crates/dbsp/src/circuit/circuit_builder.rs
@@ -62,6 +62,7 @@ use feldera_ir::{LirCircuit, LirNodeId};
 use feldera_samply::Span;
 use feldera_storage::{FileCommitter, StoragePath};
 use itertools::Itertools;
+#[cfg(unix)]
 use nix::{
     sys::time::TimeValLike,
     time::{ClockId, clock_gettime},
@@ -93,9 +94,6 @@ use tracing::debug;
 use typedmap::{TypedMap, TypedMapKey};
 
 use super::dbsp_handle::Mode;
-
-/// Label name used to store operator's persistent id,
-/// i.e., id stable across circuit modifications.
 const LABEL_PERSISTENT_OPERATOR_ID: &str = "persistent_id";
 
 /// Value stored in the stream.
@@ -8783,24 +8781,71 @@ impl<T> Timed<T> {
 pub struct ThreadCpuTime(pub Duration);
 
 impl ThreadCpuTime {
-    /// Returns the current time elapsed running the current thread.
+    /// Returns the current thread's cumulative CPU time.
     pub fn now() -> Self {
-        let nanos = clock_gettime(ClockId::CLOCK_THREAD_CPUTIME_ID)
-            .unwrap()
-            .num_nanoseconds();
-        Self(Duration::from_nanos(nanos.max(0).cast_unsigned()))
+        Self(current_thread_cpu_time())
     }
 
-    /// Returns the time elapsed running the current thread since this
-    /// `ThreadCpuTime`.
+    /// Returns the CPU time elapsed on the current thread since this
+    /// `ThreadCpuTime` was recorded.
     ///
     /// This only makes sense if this `ThreadCpuTime` was for the currently
     /// running thread.
     ///
     /// Returns zero if the current time is earlier than self.
     pub fn elapsed(&self) -> Duration {
-        Self::now().0.saturating_sub(self.0)
+        current_thread_cpu_time().saturating_sub(self.0)
+    }
+}
+
+#[cfg(unix)]
+/// Returns the current thread's cumulative CPU time from `CLOCK_THREAD_CPUTIME_ID`.
+fn current_thread_cpu_time() -> Duration {
+    let nanos = clock_gettime(ClockId::CLOCK_THREAD_CPUTIME_ID)
+        .unwrap()
+        .num_nanoseconds();
+    Duration::from_nanos(nanos.max(0).cast_unsigned())
+}
+
+#[cfg(windows)]
+/// Returns the current thread's cumulative kernel and user CPU time.
+fn current_thread_cpu_time() -> Duration {
+    use windows_sys::Win32::{
+        Foundation::FILETIME,
+        System::Threading::{GetCurrentThread, GetThreadTimes},
+    };
+
+    let mut creation = FILETIME::default();
+    let mut exit = FILETIME::default();
+    let mut kernel = FILETIME::default();
+    let mut user = FILETIME::default();
+    // SAFETY: `GetCurrentThread` returns a valid pseudo-handle, and all four
+    // initialized FILETIME values remain valid mutable output buffers for the call.
+    let result = unsafe {
+        GetThreadTimes(
+            GetCurrentThread(),
+            &mut creation,
+            &mut exit,
+            &mut kernel,
+            &mut user,
+        )
+    };
+    if result == 0 {
+        // it is very unlikely this will fail, since GetCurrentThread() always returns a 
+        // valid pseudo-handle with proper permissions
+        panic!("GetThreadTimes failed: {}", std::io::Error::last_os_error());
     }
+
+    let filetime_to_ticks =
+        |time: FILETIME| (u64::from(time.dwHighDateTime) << 32) | u64::from(time.dwLowDateTime);
+    let ticks = filetime_to_ticks(kernel).saturating_add(filetime_to_ticks(user));
+    Duration::new(ticks / 10_000_000, ((ticks % 10_000_000) * 100) as u32)
+}
+
+#[cfg(not(any(unix, windows)))]
+/// Returns zero when the platform has no supported thread CPU clock.
+fn current_thread_cpu_time() -> Duration {
+    Duration::ZERO
 }
 
 impl<T> Future for Timed<T>
@@ -8831,6 +8876,15 @@ mod tests {
     use anyhow::anyhow;
     use std::{cell::RefCell, ops::Deref, rc::Rc, vec::Vec};
 
+    #[test]
+    fn thread_cpu_time_is_monotonic() {
+        use crate::circuit::ThreadCpuTime;
+
+        let start = ThreadCpuTime::now();
+        std::hint::black_box((0..100_000u64).fold(0u64, |sum, value| sum.wrapping_add(value)));
+        assert!(ThreadCpuTime::now().0 >= start.0);
+    }
+
     #[test]
     fn sum_circuit_dynamic() {
         sum_circuit::<DynamicScheduler>();

diff --git a/crates/dbsp/src/storage/backend.rs b/crates/dbsp/src/storage/backend.rs
@@ -52,6 +52,8 @@ impl StorageCacheFlags for OpenOptions {
             use std::os::unix::fs::OpenOptionsExt;
             self.custom_flags(cache.to_custom_open_flags());
         }
+        #[cfg(not(unix))]
+        let _ = cache;
         self
     }
 }