bump rust version to 1.92 (#9609)

GitOrigin-RevId: 0a6500de567842cddfb7db4b21eadd849709386d

Author: zxqfd555

Cargo.toml

@@ -3,7 +3,7 @@ name = "pathway"
 version = "0.27.1"
 edition = "2021"
 publish = false
-rust-version = "1.88"
+rust-version = "1.92"
 license = "BUSL-1.1"
 
 [lib]

clippy.toml

@@ -1,3 +1,6 @@
+# we do not export a Rust API yet
+avoid-breaking-exported-api = false
+
 [[disallowed-methods]]
 path = "differential_dataflow::operators::Count::count"
 reason = "use crate::dataflow::operators::MaybeTotal::count instead"
@@ -45,6 +48,3 @@ reason = "use differential_dataflow::operators::join::JoinCore::join_core instea
 [[disallowed-methods]]
 path = "differential_dataflow::operators::join::Join::semijoin"
 reason = "use differential_dataflow::operators::join::JoinCore::join_core instead"
-
-# we do not export a Rust API yet
-avoid-breaking-exported-api = false

external/differential-dataflow/src/lib.rs

@@ -71,10 +71,10 @@
 //! of the new and old counts of the old and new degrees of the affected node).
 
 #![forbid(missing_docs)]
-#![allow(array_into_iter)]
 
-// Required for deriving `Abomonation`. Note: the associated crate has not been updated in 5 years and triggers warnings starting with Rust 1.85.
-#![allow(non_local_definitions)]
+// Disabling warnings since we try to keep the forks as close to the original, while the
+// suggestions are applied for the principal framework library.
+#![allow(warnings)]
 
 use std::fmt::Debug;
 

external/timely-dataflow/communication/src/lib.rs

@@ -75,8 +75,9 @@
 
 #![forbid(missing_docs)]
 
-// Required for deriving `Abomonation`. Note: the associated crate has not been updated in 5 years and triggers warnings starting with Rust 1.85.
-#![allow(non_local_definitions)]
+// Disabling warnings since we try to keep the forks as close to the original, while the
+// suggestions are applied for the principal framework library.
+#![allow(warnings)]
 
 #[cfg(feature = "getopts")]
 extern crate getopts;

external/timely-dataflow/timely/src/lib.rs

@@ -57,8 +57,9 @@
 
 #![forbid(missing_docs)]
 
-// Required for deriving `Abomonation`. Note: the associated crate has not been updated in 5 years and triggers warnings starting with Rust 1.85.
-#![allow(non_local_definitions)]
+// Disabling warnings since we try to keep the forks as close to the original, while the
+// suggestions are applied for the principal framework library.
+#![allow(warnings)]
 
 #[macro_use]
 extern crate abomonation_derive;

rust-toolchain.toml

@@ -1,2 +1,2 @@
 [toolchain]
-channel = "1.88"
+channel = "1.92"

src/connectors/adaptors.rs

@@ -44,6 +44,7 @@ pub struct UpsertSession<Timestamp: TimelyTimestamp + Lattice + TotalOrder> {
 }
 
 impl<Timestamp: TimelyTimestamp + Lattice + TotalOrder> UpsertSession<Timestamp> {
+    #[allow(clippy::wrong_self_convention)] // allowing to be consistent between implementations
     pub fn to_collection<S: MaybeTotalScope<MaybeTotalTimestamp = Timestamp>>(
         &mut self,
         scope: &mut S,

src/connectors/aws/dynamodb.rs

@@ -35,16 +35,16 @@ pub const N_SEND_ATTEMPTS: usize = 5;
 #[derive(Debug, thiserror::Error)]
 pub enum Error {
     #[error("Create table error, error details: {0:?}")]
-    CreateTableError(#[from] SdkError<CreateTableError, AwsHttpResponse>),
+    CreateTable(#[from] SdkError<CreateTableError, AwsHttpResponse>),
 
     #[error("Delete table error, error details: {0:?}")]
-    DeleteTableError(#[from] SdkError<DeleteTableError, AwsHttpResponse>),
+    DeleteTable(#[from] SdkError<DeleteTableError, AwsHttpResponse>),
 
     #[error("Describe table error, error details: {0:?}")]
-    DescribeTableError(#[from] SdkError<DescribeTableError, AwsHttpResponse>),
+    DescribeTable(#[from] SdkError<DescribeTableError, AwsHttpResponse>),
 
     #[error("Batch write error, error details: {0:?}")]
-    BatchWriteError(#[from] SdkError<BatchWriteItemError, AwsHttpResponse>),
+    BatchWrite(#[from] SdkError<BatchWriteItemError, AwsHttpResponse>),
 }
 
 pub struct DynamoDBWriter {

src/connectors/aws/kinesis.rs

@@ -32,22 +32,22 @@ use crate::retry::RetryConfig;
 #[derive(Debug, thiserror::Error)]
 pub enum Error {
     #[error("List shards error, error details: {0:?}")]
-    ListShardsError(#[from] SdkError<ListShardsError, AwsHttpResponse>),
+    ListShards(#[from] SdkError<ListShardsError, AwsHttpResponse>),
 
     #[error("List streams error, error details: {0:?}")]
-    ListStreamsError(#[from] SdkError<ListStreamsError, AwsHttpResponse>),
+    ListStreams(#[from] SdkError<ListStreamsError, AwsHttpResponse>),
 
     #[error("Describe stream error, error details: {0:?}")]
-    DescribeStreamSummaryError(#[from] SdkError<DescribeStreamSummaryError, AwsHttpResponse>),
+    DescribeStreamSummary(#[from] SdkError<DescribeStreamSummaryError, AwsHttpResponse>),
 
     #[error("Get shard iterator error, error details: {0:?}")]
-    GetShardIteratorError(#[from] SdkError<GetShardIteratorError, AwsHttpResponse>),
+    GetShardIterator(#[from] SdkError<GetShardIteratorError, AwsHttpResponse>),
 
     #[error("Get shard records error, error details: {0:?}")]
-    GetShardRecordsError(#[from] SdkError<GetRecordsError, AwsHttpResponse>),
+    GetShardRecords(#[from] SdkError<GetRecordsError, AwsHttpResponse>),
 
     #[error("Put records error, error details: {0:?}")]
-    PutRecordsError(#[from] SdkError<PutRecordsError, AwsHttpResponse>),
+    PutRecords(#[from] SdkError<PutRecordsError, AwsHttpResponse>),
 
     #[error("Iterator not found for shard {0}")]
     IteratorNotFound(String),
@@ -212,7 +212,11 @@ impl ShardSet {
                 // Update, case 1: we've read everything we could.
                 let iteration_elapsed = self.last_updated_at.elapsed();
                 if iteration_elapsed < self.round_robin_duration {
-                    std::thread::sleep(self.round_robin_duration - iteration_elapsed);
+                    std::thread::sleep(
+                        self.round_robin_duration
+                            .checked_sub(iteration_elapsed)
+                            .unwrap(),
+                    );
                 }
                 update_is_needed = true;
             } else if self.had_full_round_robin
@@ -318,7 +322,7 @@ impl KinesisReader {
         let response = iterator_builder
             .send()
             .await
-            .map_err(Error::GetShardIteratorError)?;
+            .map_err(Error::GetShardIterator)?;
         let iterator = response
             .shard_iterator()
             .ok_or(Error::IteratorNotFound(shard_id.to_string()))?
@@ -347,7 +351,7 @@ impl Reader for KinesisReader {
         let mut shard_closing_offsets: HashMap<String, String> = HashMap::new();
         for shard in &self.operated_set.assigned_shards {
             if let Some(last_record_id) = &shard.last_record_id {
-                shard_closing_offsets.insert(shard.shard_id.clone(), last_record_id.to_string());
+                shard_closing_offsets.insert(shard.shard_id.clone(), last_record_id.clone());
             }
         }
 
@@ -365,7 +369,7 @@ impl Reader for KinesisReader {
                 continue;
             };
             self.shard_offsets
-                .insert(shard_id.to_string(), offset.to_string());
+                .insert(shard_id.to_string(), offset.clone());
             if let Some(closing_offset) = shard_closing_offsets.get(shard_id.as_str()) {
                 if closing_offset == offset {
                     self.operated_set
@@ -577,7 +581,7 @@ impl KinesisWriter {
                         .stream_name(stream_name)
                         .send()
                         .await
-                        .map_err(Error::DescribeStreamSummaryError)?;
+                        .map_err(Error::DescribeStreamSummary)?;
                     let stream_status = stream
                         .stream_description_summary()
                         .ok_or(Error::StreamDoesntExist(stream_name.clone()))?
@@ -595,7 +599,7 @@ impl KinesisWriter {
                         .limit(1)
                         .send()
                         .await
-                        .map_err(Error::ListStreamsError)?;
+                        .map_err(Error::ListStreams)?;
                 }
             }
             Ok::<_, Error>(())

src/connectors/backlog.rs

@@ -37,11 +37,11 @@ impl BacklogTracker {
         }
     }
 
-    pub fn on_event(&mut self, timestamp: &Timestamp) {
+    pub fn on_event(&mut self, timestamp: Timestamp) {
         self.total_in_fly += 1;
         match self.backlog.back_mut() {
-            Some(entry) if entry.timestamp == *timestamp => entry.on_event(),
-            _ => self.backlog.push_back(BacklogEntry::new(*timestamp)),
+            Some(entry) if entry.timestamp == timestamp => entry.on_event(),
+            _ => self.backlog.push_back(BacklogEntry::new(timestamp)),
         }
     }