fix: All clippy warnings

src/common/src/array/arrow/arrow_udf.rs

@@ -124,11 +124,9 @@ impl FromArrow for UdfArrowConvert {
 
 #[cfg(test)]
 mod tests {
-    use std::sync::Arc;
 
     use super::*;
     use crate::array::*;
-    use crate::buffer::Bitmap;
 
     #[test]
     fn struct_array() {

src/connector/src/parser/avro/parser.rs

@@ -197,9 +197,7 @@ mod test {
     use crate::connector_common::AwsAuthProps;
     use crate::parser::plain_parser::PlainParser;
     use crate::parser::unified::avro::unix_epoch_days;
-    use crate::parser::{
-        AccessBuilderImpl, EncodingType, SourceStreamChunkBuilder, SpecificParserConfig,
-    };
+    use crate::parser::{AccessBuilderImpl, SourceStreamChunkBuilder, SpecificParserConfig};
     use crate::source::{SourceColumnDesc, SourceContext};
 
     fn test_data_path(file_name: &str) -> String {

src/connector/src/parser/debezium/avro_parser.rs

@@ -150,9 +150,7 @@ mod tests {
     use risingwave_pb::plan_common::{PbEncodeType, PbFormatType};
 
     use super::*;
-    use crate::parser::{
-        DebeziumAvroParserConfig, DebeziumParser, SourceStreamChunkBuilder, SpecificParserConfig,
-    };
+    use crate::parser::{DebeziumParser, SourceStreamChunkBuilder, SpecificParserConfig};
     use crate::source::{SourceColumnDesc, SourceContext};
 
     const DEBEZIUM_AVRO_DATA: &[u8] = b"\x00\x00\x00\x00\x06\x00\x02\xd2\x0f\x0a\x53\x61\x6c\x6c\x79\x0c\x54\x68\x6f\x6d\x61\x73\x2a\x73\x61\x6c\x6c\x79\x2e\x74\x68\x6f\x6d\x61\x73\x40\x61\x63\x6d\x65\x2e\x63\x6f\x6d\x16\x32\x2e\x31\x2e\x32\x2e\x46\x69\x6e\x61\x6c\x0a\x6d\x79\x73\x71\x6c\x12\x64\x62\x73\x65\x72\x76\x65\x72\x31\xc0\xb4\xe8\xb7\xc9\x61\x00\x30\x66\x69\x72\x73\x74\x5f\x69\x6e\x5f\x64\x61\x74\x61\x5f\x63\x6f\x6c\x6c\x65\x63\x74\x69\x6f\x6e\x12\x69\x6e\x76\x65\x6e\x74\x6f\x72\x79\x00\x02\x12\x63\x75\x73\x74\x6f\x6d\x65\x72\x73\x00\x00\x20\x6d\x79\x73\x71\x6c\x2d\x62\x69\x6e\x2e\x30\x30\x30\x30\x30\x33\x8c\x06\x00\x00\x00\x02\x72\x02\x92\xc3\xe8\xb7\xc9\x61\x00";

src/connector/src/source/kinesis/source/reader.rs

@@ -305,7 +305,6 @@ mod tests {
 
     use super::*;
     use crate::connector_common::KinesisCommon;
-    use crate::source::kinesis::split::KinesisSplit;
     use crate::source::SourceContext;
 
     #[tokio::test]

src/connector/src/source/nexmark/source/reader.rs

@@ -210,7 +210,7 @@ impl NexmarkSplitReader {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::source::nexmark::{NexmarkProperties, NexmarkSplitEnumerator};
+    use crate::source::nexmark::NexmarkSplitEnumerator;
     use crate::source::{SourceContext, SourceEnumeratorContext, SplitEnumerator};
 
     #[tokio::test]

src/meta/src/controller/catalog.rs

@@ -3008,7 +3008,6 @@ impl CatalogControllerInner {
 #[cfg(test)]
 #[cfg(not(madsim))]
 mod tests {
-    use risingwave_meta_model_v2::ViewId;
 
     use super::*;
 

src/storage/hummock_test/src/test_utils.rs

@@ -117,6 +117,7 @@ impl<S: LocalStateStore> TestIngestBatch for S {
 #[cfg(test)]
 #[async_trait::async_trait]
 pub(crate) trait HummockStateStoreTestTrait: StateStore {
+    #[allow(dead_code)]
     fn get_pinned_version(&self) -> PinnedVersion;
     async fn seal_and_sync_epoch(&self, epoch: u64) -> StorageResult<SyncResult> {
         self.seal_epoch(epoch, true);

src/storage/src/monitor/traced_store.rs

@@ -29,7 +29,6 @@ use crate::error::StorageResult;
 use crate::hummock::sstable_store::SstableStoreRef;
 use crate::hummock::{HummockStorage, SstableObjectIdManagerRef};
 use crate::store::*;
-use crate::StateStore;
 
 #[derive(Clone)]
 pub struct TracedStateStore<S> {

src/stream/src/common/compact_chunk.rs

@@ -333,7 +333,6 @@ pub fn merge_chunk_row(stream_chunk: StreamChunk, pk_indices: &[usize]) -> Strea
 
 #[cfg(test)]
 mod tests {
-    use risingwave_common::array::StreamChunk;
     use risingwave_common::test_prelude::StreamChunkTestExt;
 
     use super::*;

src/stream/src/executor/chain.rs

@@ -103,7 +103,6 @@ impl Execute for ChainExecutor {
 
 #[cfg(test)]
 mod test {
-    use std::default::Default;
 
     use futures::StreamExt;
     use risingwave_common::array::stream_chunk::StreamChunkTestExt;

src/stream/src/executor/dedup/append_only_dedup.rs

@@ -182,19 +182,15 @@ impl<S: StateStore> Execute for AppendOnlyDedupExecutor<S> {
 #[cfg(test)]
 mod tests {
     use std::sync::atomic::AtomicU64;
-    use std::sync::Arc;
 
-    use risingwave_common::catalog::{ColumnDesc, ColumnId, Field, Schema, TableId};
+    use risingwave_common::catalog::{ColumnDesc, ColumnId, Field, TableId};
     use risingwave_common::test_prelude::StreamChunkTestExt;
-    use risingwave_common::types::DataType;
     use risingwave_common::util::epoch::test_epoch;
     use risingwave_common::util::sort_util::OrderType;
     use risingwave_storage::memory::MemoryStateStore;
 
     use super::*;
-    use crate::common::table::state_table::StateTable;
     use crate::executor::test_utils::MockSource;
-    use crate::executor::ActorContext;
 
     #[tokio::test]
     async fn test_dedup_executor() {

src/stream/src/executor/dispatch.rs

@@ -1029,26 +1029,21 @@ impl Dispatcher for SimpleDispatcher {
 #[cfg(test)]
 mod tests {
     use std::hash::{BuildHasher, Hasher};
-    use std::sync::{Arc, Mutex};
+    use std::sync::Mutex;
 
     use async_trait::async_trait;
-    use futures::{pin_mut, StreamExt};
-    use itertools::Itertools;
+    use futures::pin_mut;
     use risingwave_common::array::stream_chunk::StreamChunkTestExt;
-    use risingwave_common::array::{Array, ArrayBuilder, I32ArrayBuilder, Op};
-    use risingwave_common::catalog::Schema;
+    use risingwave_common::array::{Array, ArrayBuilder, I32ArrayBuilder};
     use risingwave_common::config;
-    use risingwave_common::hash::VirtualNode;
     use risingwave_common::util::epoch::test_epoch;
     use risingwave_common::util::hash_util::Crc32FastBuilder;
-    use risingwave_common::util::iter_util::ZipEqFast;
     use risingwave_pb::stream_plan::DispatcherType;
 
     use super::*;
     use crate::executor::exchange::output::Output;
     use crate::executor::exchange::permit::channel_for_test;
     use crate::executor::receiver::ReceiverExecutor;
-    use crate::executor::Execute;
     use crate::task::test_utils::helper_make_local_actor;
 
     #[derive(Debug)]

src/stream/src/executor/dml.rs

@@ -273,13 +273,9 @@ impl Execute for DmlExecutor {
 
 #[cfg(test)]
 mod tests {
-    use std::sync::Arc;
 
-    use risingwave_common::array::StreamChunk;
-    use risingwave_common::catalog::{ColumnId, Field, Schema, INITIAL_TABLE_VERSION_ID};
+    use risingwave_common::catalog::{ColumnId, Field, INITIAL_TABLE_VERSION_ID};
     use risingwave_common::test_prelude::StreamChunkTestExt;
-    use risingwave_common::transaction::transaction_id::TxnId;
-    use risingwave_common::types::DataType;
     use risingwave_common::util::epoch::test_epoch;
     use risingwave_dml::dml_manager::DmlManager;
 

src/stream/src/executor/dynamic_filter.rs

@@ -488,9 +488,8 @@ impl<S: StateStore, const USE_WATERMARK_CACHE: bool> Execute
 
 #[cfg(test)]
 mod tests {
-    use risingwave_common::array::stream_chunk::StreamChunkTestExt;
     use risingwave_common::array::*;
-    use risingwave_common::catalog::{ColumnDesc, ColumnId, Field, Schema, TableId};
+    use risingwave_common::catalog::{ColumnDesc, ColumnId, Field, TableId};
     use risingwave_common::util::epoch::test_epoch;
     use risingwave_common::util::sort_util::OrderType;
     use risingwave_hummock_sdk::HummockReadEpoch;
@@ -499,7 +498,6 @@ mod tests {
 
     use super::*;
     use crate::executor::test_utils::{MessageSender, MockSource, StreamExecutorTestExt};
-    use crate::executor::{ActorContext, StreamExecutorResult};
 
     async fn create_in_memory_state_table(
         mem_state: MemoryStateStore,

src/stream/src/executor/filter.rs

@@ -157,11 +157,8 @@ impl FilterExecutor {
 
 #[cfg(test)]
 mod tests {
-    use futures::StreamExt;
     use risingwave_common::array::stream_chunk::StreamChunkTestExt;
-    use risingwave_common::array::StreamChunk;
-    use risingwave_common::catalog::{Field, Schema};
-    use risingwave_common::types::DataType;
+    use risingwave_common::catalog::Field;
 
     use super::super::test_utils::expr::build_from_pretty;
     use super::super::test_utils::MockSource;

src/stream/src/executor/hash_join.rs

@@ -1051,20 +1051,16 @@ impl<K: HashKey, S: StateStore, const T: JoinTypePrimitive> HashJoinExecutor<K,
 mod tests {
     use std::sync::atomic::AtomicU64;
 
-    use risingwave_common::array::stream_chunk::StreamChunkTestExt;
     use risingwave_common::array::*;
-    use risingwave_common::catalog::{ColumnDesc, ColumnId, Field, Schema, TableId};
+    use risingwave_common::catalog::{ColumnDesc, ColumnId, Field, TableId};
     use risingwave_common::hash::{Key128, Key64};
-    use risingwave_common::types::ScalarImpl;
     use risingwave_common::util::epoch::test_epoch;
     use risingwave_common::util::sort_util::OrderType;
     use risingwave_storage::memory::MemoryStateStore;
 
     use super::*;
-    use crate::common::table::state_table::StateTable;
     use crate::executor::test_utils::expr::build_from_pretty;
     use crate::executor::test_utils::{MessageSender, MockSource, StreamExecutorTestExt};
-    use crate::executor::{ActorContext, Barrier, EpochPair};
 
     async fn create_in_memory_state_table(
         mem_state: MemoryStateStore,

src/stream/src/executor/hop_window.rs

@@ -226,17 +226,13 @@ impl HopWindowExecutor {
 
 #[cfg(test)]
 mod tests {
-    use futures::StreamExt;
     use risingwave_common::array::stream_chunk::StreamChunkTestExt;
-    use risingwave_common::catalog::{Field, Schema};
+    use risingwave_common::catalog::Field;
     use risingwave_common::types::test_utils::IntervalTestExt;
-    use risingwave_common::types::{DataType, Interval};
     use risingwave_expr::expr::test_utils::make_hop_window_expression;
-    use risingwave_expr::expr::NonStrictExpression;
 
     use super::*;
     use crate::executor::test_utils::MockSource;
-    use crate::executor::{ActorContext, Execute, StreamChunk};
 
     const CHUNK_SIZE: usize = 256;
 

src/stream/src/executor/integration_tests.rs

@@ -12,13 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::sync::{Arc, Mutex};
+use std::sync::Mutex;
 
-use futures::StreamExt;
 use futures_async_stream::try_stream;
 use multimap::MultiMap;
 use risingwave_common::array::*;
-use risingwave_common::catalog::{Field, Schema};
+use risingwave_common::catalog::Field;
 use risingwave_common::config;
 use risingwave_common::types::*;
 use risingwave_common::util::epoch::{test_epoch, EpochExt};
@@ -29,15 +28,12 @@ use risingwave_storage::memory::MemoryStateStore;
 
 use super::exchange::permit::channel_for_test;
 use super::*;
-use crate::executor::actor::ActorContext;
 use crate::executor::dispatch::*;
 use crate::executor::exchange::output::{BoxedOutput, LocalOutput};
 use crate::executor::monitor::StreamingMetrics;
-use crate::executor::receiver::ReceiverExecutor;
 use crate::executor::test_utils::agg_executor::{
     generate_agg_schema, new_boxed_simple_agg_executor,
 };
-use crate::executor::{Execute, MergeExecutor, ProjectExecutor, StatelessSimpleAggExecutor};
 use crate::task::{LocalBarrierManager, SharedContext};
 
 /// This test creates a merger-dispatcher pair, and run a sum. Each chunk

src/stream/src/executor/join/hash_join.rs

@@ -734,7 +734,6 @@ impl JoinEntryState {
 mod tests {
     use itertools::Itertools;
     use risingwave_common::array::*;
-    use risingwave_common::types::{DataType, ScalarImpl};
     use risingwave_common::util::iter_util::ZipEqDebug;
 
     use super::*;

src/stream/src/executor/lookup_union.rs

@@ -120,10 +120,8 @@ impl LookupUnionExecutor {
 #[cfg(test)]
 mod tests {
     use futures::TryStreamExt;
-    use risingwave_common::array::StreamChunk;
-    use risingwave_common::catalog::{Field, Schema};
+    use risingwave_common::catalog::Field;
     use risingwave_common::test_prelude::StreamChunkTestExt;
-    use risingwave_common::types::DataType;
     use risingwave_common::util::epoch::test_epoch;
 
     use super::*;

src/stream/src/executor/merge.rs

@@ -412,16 +412,12 @@ impl SelectReceivers {
 
 #[cfg(test)]
 mod tests {
-    use std::collections::HashSet;
     use std::sync::atomic::{AtomicBool, Ordering};
-    use std::sync::Arc;
     use std::time::Duration;
 
     use assert_matches::assert_matches;
     use futures::FutureExt;
-    use itertools::Itertools;
-    use risingwave_common::array::{Op, StreamChunk};
-    use risingwave_common::types::ScalarImpl;
+    use risingwave_common::array::Op;
     use risingwave_common::util::epoch::test_epoch;
     use risingwave_pb::stream_plan::StreamMessage;
     use risingwave_pb::task_service::exchange_service_server::{
@@ -438,7 +434,6 @@ mod tests {
     use super::*;
     use crate::executor::exchange::input::RemoteInput;
     use crate::executor::exchange::permit::channel_for_test;
-    use crate::executor::{Barrier, Execute, Mutation};
     use crate::task::test_utils::helper_make_local_actor;
     use crate::task::LocalBarrierManager;
 

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

@@ -833,23 +833,17 @@ mod tests {
     use std::iter;
     use std::sync::atomic::AtomicU64;
 
-    use futures::stream::StreamExt;
     use rand::rngs::SmallRng;
     use rand::{Rng, RngCore, SeedableRng};
     use risingwave_common::array::stream_chunk::{StreamChunkMut, StreamChunkTestExt};
-    use risingwave_common::array::stream_chunk_builder::StreamChunkBuilder;
-    use risingwave_common::array::Op;
-    use risingwave_common::catalog::{ColumnDesc, ConflictBehavior, Field, Schema, TableId};
-    use risingwave_common::row::OwnedRow;
-    use risingwave_common::types::DataType;
+    use risingwave_common::catalog::Field;
     use risingwave_common::util::epoch::test_epoch;
-    use risingwave_common::util::sort_util::{ColumnOrder, OrderType};
+    use risingwave_common::util::sort_util::OrderType;
     use risingwave_hummock_sdk::HummockReadEpoch;
     use risingwave_storage::memory::MemoryStateStore;
     use risingwave_storage::table::batch_table::storage_table::StorageTable;
 
     use super::*;
-    use crate::executor::test_utils::prelude::StateTable;
     use crate::executor::test_utils::*;
 
     #[tokio::test]

src/stream/src/executor/project.rs

@@ -189,13 +189,11 @@ impl Inner {
 mod tests {
     use std::sync::atomic::{self, AtomicI64};
 
-    use futures::StreamExt;
     use risingwave_common::array::stream_chunk::StreamChunkTestExt;
-    use risingwave_common::array::{DataChunk, StreamChunk};
-    use risingwave_common::catalog::{Field, Schema};
-    use risingwave_common::types::{DataType, Datum};
+    use risingwave_common::array::DataChunk;
+    use risingwave_common::catalog::Field;
     use risingwave_common::util::epoch::test_epoch;
-    use risingwave_expr::expr::{self, Expression, ValueImpl};
+    use risingwave_expr::expr::{self, ValueImpl};
 
     use super::super::test_utils::MockSource;
     use super::super::*;

src/stream/src/executor/receiver.rs

@@ -191,15 +191,13 @@ impl Execute for ReceiverExecutor {
 #[cfg(test)]
 mod tests {
     use std::collections::HashMap;
-    use std::sync::Arc;
 
     use futures::{pin_mut, FutureExt};
-    use risingwave_common::array::StreamChunk;
     use risingwave_common::util::epoch::test_epoch;
     use risingwave_pb::stream_plan::update_mutation::MergeUpdate;
 
     use super::*;
-    use crate::executor::{ActorContext, Barrier, Execute, Mutation, UpdateMutation};
+    use crate::executor::UpdateMutation;
     use crate::task::test_utils::helper_make_local_actor;
 
     #[tokio::test]

src/stream/src/executor/row_id_gen.rs

@@ -123,16 +123,14 @@ impl Execute for RowIdGenExecutor {
 
 #[cfg(test)]
 mod tests {
-    use risingwave_common::array::{Array, PrimitiveArray};
-    use risingwave_common::catalog::{Field, Schema};
+    use risingwave_common::array::PrimitiveArray;
+    use risingwave_common::catalog::Field;
     use risingwave_common::hash::VirtualNode;
     use risingwave_common::test_prelude::StreamChunkTestExt;
-    use risingwave_common::types::DataType;
     use risingwave_common::util::epoch::test_epoch;
 
     use super::*;
     use crate::executor::test_utils::MockSource;
-    use crate::executor::{ActorContext, Execute};
 
     #[tokio::test]
     async fn test_row_id_gen_executor() {

src/stream/src/executor/simple_agg.rs

@@ -295,9 +295,7 @@ mod tests {
     use risingwave_common::catalog::Field;
     use risingwave_common::types::*;
     use risingwave_common::util::epoch::test_epoch;
-    use risingwave_expr::aggregate::AggCall;
     use risingwave_storage::memory::MemoryStateStore;
-    use risingwave_storage::StateStore;
 
     use super::*;
     use crate::executor::test_utils::agg_executor::new_boxed_simple_agg_executor;