fix: Replace Arc with Rc in window operators and tests to resolve Clippy warnings.

Author: argahsuknesib

CHANGELOG.md

@@ -5,7 +5,7 @@ All notable changes to the Janus project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## [Unreleased]
+## [Unreleased] - 2025-11-24
 
 ### Added
 - **Historical Sliding Window Operator** (`src/stream/operators/historical_sliding_window.rs`)
@@ -33,6 +33,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Fixed
 - Window end clamping to prevent querying future data in sliding windows
 - Type consistency across window operators
+- **Clippy warnings**: Replaced `Arc` with `Rc` in window operators and tests since they don't use threading
+  - Fixed `arc_with_non_send_sync` lint errors
+  - All CI checks now pass (rustfmt, clippy, tests, build)
 
 ## [0.1.0] - Initial Release
 

src/stream/operators/historical_fixed_window.rs

@@ -1,12 +1,12 @@
 use crate::core::Event;
 use crate::parsing::janusql_parser::WindowDefinition;
 use crate::storage::segmented_storage::StreamingSegmentedStorage;
-use std::sync::Arc;
+use std::rc::Rc;
 
 /// Operator for processing historical data with a fixed window.
 /// Unlike sliding windows, this queries a single fixed time range [start, end].
 pub struct HistoricalFixedWindowOperator {
-    storage: Arc<StreamingSegmentedStorage>,
+    storage: Rc<StreamingSegmentedStorage>,
     window_def: WindowDefinition,
     has_yielded: bool,
 }
@@ -18,7 +18,7 @@ impl HistoricalFixedWindowOperator {
     ///
     /// * `storage` - The storage backend to query.
     /// * `window_def` - The window definition with start and end timestamps.
-    pub fn new(storage: Arc<StreamingSegmentedStorage>, window_def: WindowDefinition) -> Self {
+    pub fn new(storage: Rc<StreamingSegmentedStorage>, window_def: WindowDefinition) -> Self {
         HistoricalFixedWindowOperator { storage, window_def, has_yielded: false }
     }
 }
@@ -75,7 +75,7 @@ mod tests {
         let _ = fs::remove_dir_all(test_dir);
 
         let config = create_test_config(test_dir);
-        let storage = Arc::new(StreamingSegmentedStorage::new(config).unwrap());
+        let storage = Rc::new(StreamingSegmentedStorage::new(config).unwrap());
 
         // Write events at timestamps 100, 200, 300, 400, 500, 600
         for i in 1..=6 {

src/stream/operators/historical_sliding_window.rs

@@ -1,12 +1,12 @@
 use crate::core::Event;
 use crate::parsing::janusql_parser::WindowDefinition;
 use crate::storage::segmented_storage::StreamingSegmentedStorage;
-use std::sync::Arc;
+use std::rc::Rc;
 
 /// Operator for processing historical data with a sliding window.
 /// It iterates over the storage and yields events for each window.
 pub struct HistoricalSlidingWindowOperator {
-    storage: Arc<StreamingSegmentedStorage>,
+    storage: Rc<StreamingSegmentedStorage>,
     window_def: WindowDefinition,
     current_start: u64,
     end_bound: u64,
@@ -19,7 +19,7 @@ impl HistoricalSlidingWindowOperator {
     ///
     /// * `storage` - The storage backend to query.
     /// * `window_def` - The window definition (width, slide, offset, etc.).
-    pub fn new(storage: Arc<StreamingSegmentedStorage>, window_def: WindowDefinition) -> Self {
+    pub fn new(storage: Rc<StreamingSegmentedStorage>, window_def: WindowDefinition) -> Self {
         let now = std::time::SystemTime::now()
             .duration_since(std::time::UNIX_EPOCH)
             .unwrap()
@@ -98,7 +98,7 @@ mod tests {
         let _ = fs::remove_dir_all(test_dir); // Clean up before test
 
         let config = create_test_config(test_dir);
-        let storage = Arc::new(StreamingSegmentedStorage::new(config).unwrap());
+        let storage = Rc::new(StreamingSegmentedStorage::new(config).unwrap());
 
         let now = std::time::SystemTime::now()
             .duration_since(std::time::UNIX_EPOCH)

tests/historical_fixed_window_test.rs

@@ -3,7 +3,7 @@ use janus::storage::segmented_storage::StreamingSegmentedStorage;
 use janus::storage::util::StreamingConfig;
 use janus::stream::operators::historical_fixed_window::HistoricalFixedWindowOperator;
 use std::fs;
-use std::sync::Arc;
+use std::rc::Rc;
 
 fn create_test_config(path: &str) -> StreamingConfig {
     StreamingConfig {
@@ -22,7 +22,7 @@ fn test_historical_fixed_window_with_real_iris() {
     let _ = fs::remove_dir_all(test_dir);
 
     let config = create_test_config(test_dir);
-    let storage = Arc::new(StreamingSegmentedStorage::new(config).unwrap());
+    let storage = Rc::new(StreamingSegmentedStorage::new(config).unwrap());
 
     // Write events with real RDF IRIs representing IoT device events
     let devices = [
@@ -107,7 +107,7 @@ fn test_historical_fixed_window_semantic_web() {
     let _ = fs::remove_dir_all(test_dir);
 
     let config = create_test_config(test_dir);
-    let storage = Arc::new(StreamingSegmentedStorage::new(config).unwrap());
+    let storage = Rc::new(StreamingSegmentedStorage::new(config).unwrap());
 
     // Semantic web example: Publications and authors
     let publications = [

tests/historical_sliding_window_test.rs

@@ -3,7 +3,7 @@ use janus::storage::segmented_storage::StreamingSegmentedStorage;
 use janus::storage::util::StreamingConfig;
 use janus::stream::operators::historical_sliding_window::HistoricalSlidingWindowOperator;
 use std::fs;
-use std::sync::Arc;
+use std::rc::Rc;
 
 fn create_test_config(path: &str) -> StreamingConfig {
     StreamingConfig {
@@ -22,7 +22,7 @@ fn test_historical_sliding_window_with_real_iris() {
     let _ = fs::remove_dir_all(test_dir);
 
     let config = create_test_config(test_dir);
-    let storage = Arc::new(StreamingSegmentedStorage::new(config).unwrap());
+    let storage = Rc::new(StreamingSegmentedStorage::new(config).unwrap());
 
     let now = std::time::SystemTime::now()
         .duration_since(std::time::UNIX_EPOCH)
@@ -102,7 +102,7 @@ fn test_historical_sliding_window_foaf_example() {
     let _ = fs::remove_dir_all(test_dir);
 
     let config = create_test_config(test_dir);
-    let storage = Arc::new(StreamingSegmentedStorage::new(config).unwrap());
+    let storage = Rc::new(StreamingSegmentedStorage::new(config).unwrap());
 
     let now = std::time::SystemTime::now()
         .duration_since(std::time::UNIX_EPOCH)