fix lints

Signed-off-by: xxchan <[email protected]>

Cargo.toml

@@ -253,6 +253,8 @@ ptr_arg = "allow"
 get_first = "allow"
 # https://github.com/rust-lang/rust-clippy/issues/12016
 blocks_in_conditions = "allow"
+# https://github.com/rust-lang/rust-clippy/issues/12537
+duplicated_attributes = "allow"
 
 [workspace.lints.rustdoc]
 private_intra_doc_links = "allow"

src/batch/src/executor/join/hash_join.rs

@@ -580,7 +580,7 @@ impl<K: HashKey> HashJoinExecutor<K> {
                 }
                 shutdown_rx.check()?;
                 if !ANTI_JOIN {
-                    if hash_map.get(probe_key).is_some() {
+                    if hash_map.contains_key(probe_key) {
                         if let Some(spilled) = Self::append_one_probe_row(
                             &mut chunk_builder,
                             &probe_chunk,

src/compute/src/telemetry.rs

@@ -33,6 +33,7 @@ impl ComputeTelemetryCreator {
 
 #[async_trait::async_trait]
 impl TelemetryReportCreator for ComputeTelemetryCreator {
+    #[expect(refining_impl_trait)]
     async fn create_report(
         &self,
         tracking_id: String,

src/connector/src/sink/encoder/avro.rs

@@ -526,7 +526,7 @@ mod tests {
 
         test_ok(
             &DataType::Int64,
-            Some(ScalarImpl::Int64(std::i64::MAX)),
+            Some(ScalarImpl::Int64(i64::MAX)),
             r#""long""#,
             Value::Long(i64::MAX),
         );

src/connector/src/sink/encoder/json.rs

@@ -508,7 +508,7 @@ mod tests {
                 data_type: DataType::Int64,
                 ..mock_field.clone()
             },
-            Some(ScalarImpl::Int64(std::i64::MAX).as_scalar_ref_impl()),
+            Some(ScalarImpl::Int64(i64::MAX).as_scalar_ref_impl()),
             DateHandlingMode::FromCe,
             TimestampHandlingMode::String,
             TimestamptzHandlingMode::UtcString,
@@ -518,7 +518,7 @@ mod tests {
         .unwrap();
         assert_eq!(
             serde_json::to_string(&int64_value).unwrap(),
-            std::i64::MAX.to_string()
+            i64::MAX.to_string()
         );
 
         // https://github.com/debezium/debezium/blob/main/debezium-core/src/main/java/io/debezium/time/ZonedTimestamp.java

src/connector/src/source/filesystem/opendal_source/opendal_enumerator.rs

@@ -56,10 +56,7 @@ impl<Src: OpendalSource> SplitEnumerator for OpendalEnumerator<Src> {
 
 impl<Src: OpendalSource> OpendalEnumerator<Src> {
     pub async fn list(&self) -> ConnectorResult<ObjectMetadataIter> {
-        let prefix = match &self.prefix {
-            Some(prefix) => prefix,
-            None => "",
-        };
+        let prefix = self.prefix.as_deref().unwrap_or("");
 
         let object_lister = self
             .op

src/frontend/src/binder/bind_context.rs

@@ -107,6 +107,7 @@ pub enum BindingCteState {
 pub struct RecursiveUnion {
     /// currently this *must* be true,
     /// otherwise binding will fail.
+    #[expect(dead_code)]
     pub all: bool,
     /// lhs part of the `UNION ALL` operator
     pub base: Box<BoundSetExpr>,

src/frontend/src/binder/relation/join.rs

@@ -164,10 +164,10 @@ impl Binder {
                     JoinConstraint::Using(cols) => {
                         // sanity check
                         for col in &cols {
-                            if old_context.indices_of.get(&col.real_value()).is_none() {
+                            if !old_context.indices_of.contains_key(&col.real_value()) {
                                 return Err(ErrorCode::ItemNotFound(format!("column \"{}\" specified in USING clause does not exist in left table", col.real_value())).into());
                             }
-                            if self.context.indices_of.get(&col.real_value()).is_none() {
+                            if !self.context.indices_of.contains_key(&col.real_value()) {
                                 return Err(ErrorCode::ItemNotFound(format!("column \"{}\" specified in USING clause does not exist in right table", col.real_value())).into());
                             }
                         }

src/frontend/src/catalog/system_catalog/rw_catalog/rw_iceberg_files.rs

@@ -44,7 +44,7 @@ struct RwIcebergFiles {
     /// Total file size in bytes
     pub file_size_in_bytes: i64,
     /// Field ids used to determine row equality in equality delete files.
-    /// Required when content is EqualityDeletes and should be null
+    /// Required when content is `EqualityDeletes` and should be null
     /// otherwise. Fields with ids listed in this column must be present
     /// in the delete file
     pub equality_ids: Option<Vec<i32>>,

src/frontend/src/handler/create_sink.rs

@@ -553,7 +553,7 @@ fn check_cycle_for_sink(
                     path.push(table.name.clone());
                     self.visit_table(table.as_ref(), target_table_id, path)?;
                     path.pop();
-                } else if self.source_index.get(&table_id.table_id).is_some() {
+                } else if self.source_index.contains_key(&table_id.table_id) {
                     continue;
                 } else {
                     bail!("streaming job not found: {:?}", table_id);

src/frontend/src/handler/create_source.rs

@@ -1614,7 +1614,7 @@ pub mod tests {
         );
 
         // Options are not merged into props.
-        assert!(source.with_properties.get("schema.location").is_none());
+        assert!(!source.with_properties.contains_key("schema.location"));
     }
 
     #[tokio::test]

src/frontend/src/handler/create_table.rs

@@ -1415,6 +1415,6 @@ mod tests {
         );
 
         // Options are not merged into props.
-        assert!(source.with_properties.get("schema.location").is_none());
+        assert!(!source.with_properties.contains_key("schema.location"));
     }
 }

src/frontend/src/optimizer/plan_node/generic/project.rs

@@ -86,7 +86,7 @@ impl<PlanRef: GenericPlanRef> GenericPlanNode for Project<PlanRef> {
                     Some(input_idx) => {
                         let mut field = input_schema.fields()[input_idx].clone();
                         if let Some(name) = self.field_names.get(&i) {
-                            field.name = name.clone();
+                            field.name.clone_from(name);
                         }
                         (field.name, field.sub_fields, field.type_name)
                     }

src/frontend/src/scheduler/distributed/stage.rs

@@ -93,6 +93,7 @@ pub enum StageEvent {
         reason: SchedulerError,
     },
     /// All tasks in stage finished.
+    #[expect(dead_code)]
     Completed(StageId),
 }
 

src/frontend/src/telemetry.rs

@@ -33,6 +33,7 @@ impl FrontendTelemetryCreator {
 
 #[async_trait::async_trait]
 impl TelemetryReportCreator for FrontendTelemetryCreator {
+    #[expect(refining_impl_trait)]
     async fn create_report(
         &self,
         tracking_id: String,

src/meta/src/barrier/command.rs

@@ -323,7 +323,7 @@ pub struct CommandContext {
     /// Differs from [`TracedEpoch`], this span focuses on the lifetime of the corresponding
     /// barrier, including the process of waiting for the barrier to be sent, flowing through the
     /// stream graph on compute nodes, and finishing its `post_collect` stuffs.
-    pub span: tracing::Span,
+    pub _span: tracing::Span,
 }
 
 impl CommandContext {
@@ -346,7 +346,7 @@ impl CommandContext {
             command,
             kind,
             barrier_manager_context,
-            span,
+            _span: span,
         }
     }
 

src/meta/src/barrier/mod.rs

@@ -422,6 +422,7 @@ enum CompletingCommand {
         // that has finished but not checkpointed. If there is any, we will force checkpoint on the next barrier
         join_handle: JoinHandle<MetaResult<BarrierCompleteOutput>>,
     },
+    #[expect(dead_code)]
     Err(MetaError),
 }
 

src/meta/src/hummock/compaction/picker/mod.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#![expect(dead_code)]
+
 mod base_level_compaction_picker;
 mod emergency_compaction_picker;
 mod intra_compaction_picker;

src/meta/src/hummock/compaction/picker/space_reclaim_compaction_picker.rs

@@ -24,7 +24,7 @@ use crate::hummock::level_handler::LevelHandler;
 // key_range and selects the appropriate files to generate compaction
 pub struct SpaceReclaimCompactionPicker {
     // config
-    pub max_space_reclaim_bytes: u64,
+    pub _max_space_reclaim_bytes: u64,
 
     // for filter
     pub all_table_ids: HashSet<u32>,
@@ -40,7 +40,7 @@ pub struct SpaceReclaimPickerState {
 impl SpaceReclaimCompactionPicker {
     pub fn new(max_space_reclaim_bytes: u64, all_table_ids: HashSet<u32>) -> Self {
         Self {
-            max_space_reclaim_bytes,
+            _max_space_reclaim_bytes: max_space_reclaim_bytes,
             all_table_ids,
         }
     }

src/meta/src/hummock/manager/compaction.rs

@@ -47,7 +47,7 @@ pub struct Compaction {
     /// `CompactStatus` of each compaction group
     pub compaction_statuses: BTreeMap<CompactionGroupId, CompactStatus>,
 
-    pub deterministic_mode: bool,
+    pub _deterministic_mode: bool,
 }
 
 impl HummockManager {

src/meta/src/hummock/manager/mod.rs

@@ -919,7 +919,7 @@ impl HummockManager {
                 break;
             }
 
-            if current_version.levels.get(&compaction_group_id).is_none() {
+            if !current_version.levels.contains_key(&compaction_group_id) {
                 continue;
             }
 
@@ -1051,8 +1051,9 @@ impl HummockManager {
                     compact_task.set_task_status(TaskStatus::Success);
                     compact_status.report_compact_task(&compact_task);
                     if !is_trivial_reclaim {
-                        compact_task.sorted_output_ssts =
-                            compact_task.input_ssts[0].table_infos.clone();
+                        compact_task
+                            .sorted_output_ssts
+                            .clone_from(&compact_task.input_ssts[0].table_infos);
                     }
                     self.metrics
                         .compact_frequency

src/meta/src/manager/catalog/mod.rs

@@ -3956,7 +3956,7 @@ impl CatalogManager {
                 id
             )));
         }
-        if user_core.catalog_create_ref_count.get(&id).is_some() {
+        if user_core.catalog_create_ref_count.contains_key(&id) {
             return Err(MetaError::permission_denied(format!(
                 "User {} cannot be dropped because some objects depend on it",
                 user.name

src/meta/src/manager/metadata.rs

@@ -63,6 +63,7 @@ pub struct MetadataManagerV2 {
 #[derive(Debug)]
 pub(crate) enum ActiveStreamingWorkerChange {
     Add(WorkerNode),
+    #[expect(dead_code)]
     Remove(WorkerNode),
     Update(WorkerNode),
 }

src/sqlparser/tests/test_utils/mod.rs

@@ -11,6 +11,7 @@
 // limitations under the License.
 
 // Re-export everything from `src/test_utils.rs`.
+#[allow(unused_imports)]
 pub use risingwave_sqlparser::test_utils::*;
 
 // For the test-only macros we take a different approach of keeping them here

src/storage/compactor/src/lib.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#![feature(lint_reasons)]
+
 mod compactor_observer;
 mod rpc;
 pub mod server;

src/storage/compactor/src/telemetry.rs

@@ -33,6 +33,7 @@ impl CompactorTelemetryCreator {
 
 #[async_trait::async_trait]
 impl TelemetryReportCreator for CompactorTelemetryCreator {
+    #[expect(refining_impl_trait)]
     async fn create_report(
         &self,
         tracking_id: String,

src/stream/src/common/compact_chunk.rs

@@ -105,7 +105,7 @@ type OpRowMap<'a, 'b> =
 pub enum RowOp<'a> {
     Insert(RowRef<'a>),
     Delete(RowRef<'a>),
-    /// (old_value, new_value)
+    /// (`old_value`, `new_value`)
     Update((RowRef<'a>, RowRef<'a>)),
 }
 static LOG_SUPPERSSER: LazyLock<LogSuppresser> = LazyLock::new(LogSuppresser::default);

src/stream/src/executor/backfill/cdc/upstream_table/snapshot.rs

@@ -40,18 +40,18 @@ pub trait UpstreamTableRead {
 
 #[derive(Debug, Default)]
 pub struct SnapshotReadArgs {
-    pub epoch: u64,
+    pub _epoch: u64,
     pub current_pos: Option<OwnedRow>,
-    pub ordered: bool,
+    pub _ordered: bool,
     pub chunk_size: usize,
 }
 
 impl SnapshotReadArgs {
     pub fn new_for_cdc(current_pos: Option<OwnedRow>, chunk_size: usize) -> Self {
         Self {
-            epoch: INVALID_EPOCH,
+            _epoch: INVALID_EPOCH,
             current_pos,
-            ordered: false,
+            _ordered: false,
             chunk_size,
         }
     }

src/stream/src/executor/mview/materialize.rs

@@ -266,7 +266,7 @@ impl<S: StateStore, SD: ValueRowSerde> MaterializeExecutor<S, SD> {
         };
         match mutation {
             // Add is for mv, index and sink creation.
-            Mutation::Add(AddMutation { adds, .. }) => adds.get(&actor_id).is_some(),
+            Mutation::Add(AddMutation { adds, .. }) => adds.contains_key(&actor_id),
             // AddAndUpdate is for sink-into-table.
             Mutation::AddAndUpdate(
                 AddMutation { adds, .. },
@@ -276,9 +276,9 @@ impl<S: StateStore, SD: ValueRowSerde> MaterializeExecutor<S, SD> {
                     ..
                 },
             ) => {
-                adds.get(&actor_id).is_some()
-                    || actor_dispatchers.get(&actor_id).is_some()
-                    || dispatchers.get(&actor_id).is_some()
+                adds.contains_key(&actor_id)
+                    || actor_dispatchers.contains_key(&actor_id)
+                    || dispatchers.contains_key(&actor_id)
             }
             Mutation::Update(_)
             | Mutation::Stop(_)

src/stream/src/executor/source/source_backfill_executor.rs

@@ -730,11 +730,11 @@ impl<S: StateStore> SourceBackfillExecutorInner<S> {
         if split_changed {
             stage
                 .unfinished_splits
-                .retain(|split| target_state.get(split.id().as_ref()).is_some());
+                .retain(|split| target_state.contains_key(split.id().as_ref()));
 
             let dropped_splits = stage
                 .states
-                .extract_if(|split_id, _| target_state.get(split_id).is_none())
+                .extract_if(|split_id, _| !target_state.contains_key(split_id))
                 .map(|(split_id, _)| split_id);
 
             if should_trim_state {
@@ -823,7 +823,7 @@ impl<S: StateStore> SourceBackfillExecutorInner<S> {
             );
 
             let dropped_splits =
-                current_splits.extract_if(|split_id| target_splits.get(split_id).is_none());
+                current_splits.extract_if(|split_id| !target_splits.contains(split_id));
 
             if should_trim_state {
                 // trim dropped splits' state

src/stream/src/executor/source/source_executor.rs

@@ -226,11 +226,11 @@ impl<S: StateStore> SourceExecutor<S> {
             );
 
             core.updated_splits_in_epoch
-                .retain(|split_id, _| target_state.get(split_id).is_some());
+                .retain(|split_id, _| target_state.contains_key(split_id));
 
             let dropped_splits = core
                 .latest_split_info
-                .extract_if(|split_id, _| target_state.get(split_id).is_none())
+                .extract_if(|split_id, _| !target_state.contains_key(split_id))
                 .map(|(_, split)| split)
                 .collect_vec();
 

src/tests/compaction_test/src/lib.rs

@@ -44,7 +44,7 @@ pub struct CompactionTestOpts {
     #[clap(long)]
     pub client_address: Option<String>,
 
-    /// The state store string e.g. hummock+s3://test-bucket
+    /// The state store string e.g. `hummock+s3://test-bucket`
     #[clap(short, long)]
     pub state_store: String,