diff --git a/src/common/src/array/arrow/arrow_udf.rs b/src/common/src/array/arrow/arrow_udf.rs
index 5a44ef1439619..17425d8449a52 100644
--- a/src/common/src/array/arrow/arrow_udf.rs
+++ b/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() {
diff --git a/src/connector/src/parser/avro/parser.rs b/src/connector/src/parser/avro/parser.rs
index 9ec35a02197bf..c1950313b5e5c 100644
--- a/src/connector/src/parser/avro/parser.rs
+++ b/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 {
diff --git a/src/connector/src/parser/debezium/avro_parser.rs b/src/connector/src/parser/debezium/avro_parser.rs
index f74e1289bf007..3b11a1926f107 100644
--- a/src/connector/src/parser/debezium/avro_parser.rs
+++ b/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";
diff --git a/src/connector/src/source/kinesis/source/reader.rs b/src/connector/src/source/kinesis/source/reader.rs
index 9a72998a341f6..4504324c5fb96 100644
--- a/src/connector/src/source/kinesis/source/reader.rs
+++ b/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]
diff --git a/src/connector/src/source/nexmark/source/reader.rs b/src/connector/src/source/nexmark/source/reader.rs
index 6441baa154ae4..d1f10f4a00fc6 100644
--- a/src/connector/src/source/nexmark/source/reader.rs
+++ b/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]
diff --git a/src/meta/src/controller/catalog.rs b/src/meta/src/controller/catalog.rs
index 409672138d79f..f8ec470a74918 100644
--- a/src/meta/src/controller/catalog.rs
+++ b/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::*;
 
diff --git a/src/storage/hummock_test/src/test_utils.rs b/src/storage/hummock_test/src/test_utils.rs
index 06a0d43ccac83..9a4e674e1f8a9 100644
--- a/src/storage/hummock_test/src/test_utils.rs
+++ b/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);
diff --git a/src/storage/src/monitor/traced_store.rs b/src/storage/src/monitor/traced_store.rs
index 47c0de67729b6..df93f4c2d1a12 100644
--- a/src/storage/src/monitor/traced_store.rs
+++ b/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> {
diff --git a/src/stream/src/common/compact_chunk.rs b/src/stream/src/common/compact_chunk.rs
index dc5ae1561a2ba..5d6acb5125422 100644
--- a/src/stream/src/common/compact_chunk.rs
+++ b/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::*;
diff --git a/src/stream/src/executor/chain.rs b/src/stream/src/executor/chain.rs
index ebcbe1e4e49bb..96b9422a97b2c 100644
--- a/src/stream/src/executor/chain.rs
+++ b/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;
diff --git a/src/stream/src/executor/dedup/append_only_dedup.rs b/src/stream/src/executor/dedup/append_only_dedup.rs
index f73e196815400..f98847cf1de4b 100644
--- a/src/stream/src/executor/dedup/append_only_dedup.rs
+++ b/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() {
diff --git a/src/stream/src/executor/dispatch.rs b/src/stream/src/executor/dispatch.rs
index 6c644466f9683..9e6503c2d9a28 100644
--- a/src/stream/src/executor/dispatch.rs
+++ b/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)]
diff --git a/src/stream/src/executor/dml.rs b/src/stream/src/executor/dml.rs
index b8839d76000c1..adcb3f01ab8bd 100644
--- a/src/stream/src/executor/dml.rs
+++ b/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;
 
diff --git a/src/stream/src/executor/dynamic_filter.rs b/src/stream/src/executor/dynamic_filter.rs
index eb2a0648225d3..e9c631202e8bb 100644
--- a/src/stream/src/executor/dynamic_filter.rs
+++ b/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,
diff --git a/src/stream/src/executor/filter.rs b/src/stream/src/executor/filter.rs
index 4d1ecb098bd8f..b183719a9a677 100644
--- a/src/stream/src/executor/filter.rs
+++ b/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;
diff --git a/src/stream/src/executor/hash_join.rs b/src/stream/src/executor/hash_join.rs
index a06d490400cdc..60e2e532eb1fa 100644
--- a/src/stream/src/executor/hash_join.rs
+++ b/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,
diff --git a/src/stream/src/executor/hop_window.rs b/src/stream/src/executor/hop_window.rs
index 28408ad046a4e..cd3c0071a18c7 100644
--- a/src/stream/src/executor/hop_window.rs
+++ b/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;
 
diff --git a/src/stream/src/executor/integration_tests.rs b/src/stream/src/executor/integration_tests.rs
index d29224aa63aa4..e341adad9df5c 100644
--- a/src/stream/src/executor/integration_tests.rs
+++ b/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
diff --git a/src/stream/src/executor/join/hash_join.rs b/src/stream/src/executor/join/hash_join.rs
index 88b3ede24b619..8805ea3cc3120 100644
--- a/src/stream/src/executor/join/hash_join.rs
+++ b/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::*;
diff --git a/src/stream/src/executor/lookup_union.rs b/src/stream/src/executor/lookup_union.rs
index 62f3a0cfcb350..8c0e4103b5b2c 100644
--- a/src/stream/src/executor/lookup_union.rs
+++ b/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::*;
diff --git a/src/stream/src/executor/merge.rs b/src/stream/src/executor/merge.rs
index 96d9826be5994..6d3f81928a302 100644
--- a/src/stream/src/executor/merge.rs
+++ b/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;
 
diff --git a/src/stream/src/executor/mview/materialize.rs b/src/stream/src/executor/mview/materialize.rs
index 7aff6e33745dc..9ad0dfbdfcea3 100644
--- a/src/stream/src/executor/mview/materialize.rs
+++ b/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]
diff --git a/src/stream/src/executor/project.rs b/src/stream/src/executor/project.rs
index e78238585c9fa..22ce33a1066d5 100644
--- a/src/stream/src/executor/project.rs
+++ b/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::*;
diff --git a/src/stream/src/executor/receiver.rs b/src/stream/src/executor/receiver.rs
index 3966b25febf5d..1493e3ce17cf9 100644
--- a/src/stream/src/executor/receiver.rs
+++ b/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]
diff --git a/src/stream/src/executor/row_id_gen.rs b/src/stream/src/executor/row_id_gen.rs
index c779092a307fb..2da75389fd9f7 100644
--- a/src/stream/src/executor/row_id_gen.rs
+++ b/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() {
diff --git a/src/stream/src/executor/simple_agg.rs b/src/stream/src/executor/simple_agg.rs
index 9c35627857828..e30ceaf73a851 100644
--- a/src/stream/src/executor/simple_agg.rs
+++ b/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;
diff --git a/src/stream/src/executor/sink.rs b/src/stream/src/executor/sink.rs
index 1de9f65e19c1b..e12590527e3a8 100644
--- a/src/stream/src/executor/sink.rs
+++ b/src/stream/src/executor/sink.rs
@@ -480,7 +480,6 @@ mod test {
     use super::*;
     use crate::common::log_store_impl::in_mem::BoundedInMemLogStoreFactory;
     use crate::executor::test_utils::*;
-    use crate::executor::ActorContext;
 
     #[tokio::test]
     async fn test_force_append_only_sink() {
diff --git a/src/stream/src/executor/sort.rs b/src/stream/src/executor/sort.rs
index 37d08c746bba8..c2a01c8915a96 100644
--- a/src/stream/src/executor/sort.rs
+++ b/src/stream/src/executor/sort.rs
@@ -143,16 +143,13 @@ impl<S: StateStore> SortExecutor<S> {
 #[cfg(test)]
 mod tests {
     use risingwave_common::array::stream_chunk::StreamChunkTestExt;
-    use risingwave_common::array::StreamChunk;
-    use risingwave_common::catalog::{ColumnDesc, ColumnId, Field, Schema, TableId};
-    use risingwave_common::types::DataType;
+    use risingwave_common::catalog::{ColumnDesc, ColumnId, Field, TableId};
     use risingwave_common::util::epoch::test_epoch;
     use risingwave_common::util::sort_util::OrderType;
     use risingwave_storage::memory::MemoryStateStore;
 
     use super::*;
     use crate::executor::test_utils::{MessageSender, MockSource, StreamExecutorTestExt};
-    use crate::executor::{ActorContext, BoxedMessageStream, Execute};
 
     async fn create_executor<S: StateStore>(
         sort_column_index: usize,
diff --git a/src/stream/src/executor/source/source_executor.rs b/src/stream/src/executor/source/source_executor.rs
index a793596c8ff67..a53f48827454f 100644
--- a/src/stream/src/executor/source/source_executor.rs
+++ b/src/stream/src/executor/source/source_executor.rs
@@ -766,15 +766,11 @@ impl<S: StateStore> WaitEpochWorker<S> {
 #[cfg(test)]
 mod tests {
     use std::collections::HashSet;
-    use std::time::Duration;
 
-    use futures::StreamExt;
     use maplit::{convert_args, hashmap};
-    use risingwave_common::array::StreamChunk;
-    use risingwave_common::catalog::{ColumnId, Field, Schema, TableId};
+    use risingwave_common::catalog::{ColumnId, Field, TableId};
     use risingwave_common::system_param::local_manager::LocalSystemParamsManager;
     use risingwave_common::test_prelude::StreamChunkTestExt;
-    use risingwave_common::types::DataType;
     use risingwave_common::util::epoch::test_epoch;
     use risingwave_connector::source::datagen::DatagenSplit;
     use risingwave_connector::source::reader::desc::test_utils::create_source_desc_builder;
@@ -786,7 +782,6 @@ mod tests {
 
     use super::*;
     use crate::executor::source::{default_source_internal_table, SourceStateTableHandler};
-    use crate::executor::ActorContext;
 
     const MOCK_SOURCE_NAME: &str = "mock_source";
 
diff --git a/src/stream/src/executor/source/state_table_handler.rs b/src/stream/src/executor/source/state_table_handler.rs
index 7fc12000656cd..1d14de9a493c9 100644
--- a/src/stream/src/executor/source/state_table_handler.rs
+++ b/src/stream/src/executor/source/state_table_handler.rs
@@ -257,11 +257,9 @@ pub fn default_source_internal_table(id: u32) -> PbTable {
 
 #[cfg(test)]
 pub(crate) mod tests {
-    use std::sync::Arc;
 
-    use risingwave_common::row::OwnedRow;
-    use risingwave_common::types::{Datum, ScalarImpl};
-    use risingwave_common::util::epoch::{test_epoch, EpochPair};
+    use risingwave_common::types::Datum;
+    use risingwave_common::util::epoch::test_epoch;
     use risingwave_connector::source::kafka::KafkaSplit;
     use risingwave_storage::memory::MemoryStateStore;
     use serde_json::Value;
diff --git a/src/stream/src/executor/stateless_simple_agg.rs b/src/stream/src/executor/stateless_simple_agg.rs
index 1360f4166008e..dd1f39580c367 100644
--- a/src/stream/src/executor/stateless_simple_agg.rs
+++ b/src/stream/src/executor/stateless_simple_agg.rs
@@ -125,16 +125,13 @@ impl StatelessSimpleAggExecutor {
 #[cfg(test)]
 mod tests {
     use assert_matches::assert_matches;
-    use futures::StreamExt;
     use risingwave_common::array::stream_chunk::StreamChunkTestExt;
-    use risingwave_common::array::StreamChunk;
     use risingwave_common::catalog::schema_test_utils;
     use risingwave_common::util::epoch::test_epoch;
 
     use super::*;
     use crate::executor::test_utils::agg_executor::generate_agg_schema;
     use crate::executor::test_utils::MockSource;
-    use crate::executor::{Execute, StatelessSimpleAggExecutor};
 
     #[tokio::test]
     async fn test_no_chunk() {
diff --git a/src/stream/src/executor/top_n/group_top_n.rs b/src/stream/src/executor/top_n/group_top_n.rs
index 3a8a618e3c836..d6b2af690d838 100644
--- a/src/stream/src/executor/top_n/group_top_n.rs
+++ b/src/stream/src/executor/top_n/group_top_n.rs
@@ -267,11 +267,9 @@ mod tests {
     use std::sync::atomic::AtomicU64;
 
     use assert_matches::assert_matches;
-    use futures::StreamExt;
     use risingwave_common::array::stream_chunk::StreamChunkTestExt;
-    use risingwave_common::catalog::{Field, Schema};
+    use risingwave_common::catalog::Field;
     use risingwave_common::hash::SerializedKey;
-    use risingwave_common::types::DataType;
     use risingwave_common::util::epoch::test_epoch;
     use risingwave_common::util::sort_util::OrderType;
     use risingwave_storage::memory::MemoryStateStore;
@@ -279,7 +277,6 @@ mod tests {
     use super::*;
     use crate::executor::test_utils::top_n_executor::create_in_memory_state_table;
     use crate::executor::test_utils::MockSource;
-    use crate::executor::{ActorContext, Barrier, Execute, Message};
 
     fn create_schema() -> Schema {
         Schema {
diff --git a/src/stream/src/executor/top_n/top_n_plain.rs b/src/stream/src/executor/top_n/top_n_plain.rs
index 30aee860a8fc8..25227b5f556a0 100644
--- a/src/stream/src/executor/top_n/top_n_plain.rs
+++ b/src/stream/src/executor/top_n/top_n_plain.rs
@@ -185,7 +185,6 @@ where
 #[cfg(test)]
 mod tests {
     use assert_matches::assert_matches;
-    use futures::StreamExt;
     use risingwave_common::array::stream_chunk::StreamChunkTestExt;
     use risingwave_common::catalog::{Field, Schema};
     use risingwave_common::types::DataType;
@@ -201,7 +200,6 @@ mod tests {
         use risingwave_common::util::epoch::test_epoch;
 
         use super::*;
-        use crate::executor::{ActorContext, Execute};
         fn create_stream_chunks() -> Vec<StreamChunk> {
             let chunk1 = StreamChunk::from_pretty(
                 "  I I
@@ -685,7 +683,6 @@ mod tests {
 
         use super::*;
         use crate::executor::test_utils::top_n_executor::create_in_memory_state_table_from_state_store;
-        use crate::executor::{ActorContext, Execute};
 
         fn create_source_new() -> Executor {
             let mut chunks = vec![
@@ -1002,7 +999,6 @@ mod tests {
 
         use super::*;
         use crate::executor::test_utils::top_n_executor::create_in_memory_state_table_from_state_store;
-        use crate::executor::{ActorContext, Execute};
 
         fn create_source() -> Executor {
             let mut chunks = vec![
diff --git a/src/stream/src/executor/union.rs b/src/stream/src/executor/union.rs
index ac8f3581dda18..f72f5170ce868 100644
--- a/src/stream/src/executor/union.rs
+++ b/src/stream/src/executor/union.rs
@@ -145,12 +145,9 @@ async fn merge(inputs: Vec<BoxedMessageStream>) {
 mod tests {
     use async_stream::try_stream;
     use risingwave_common::array::stream_chunk::StreamChunkTestExt;
-    use risingwave_common::array::StreamChunk;
-    use risingwave_common::types::{DataType, ScalarImpl};
     use risingwave_common::util::epoch::test_epoch;
 
     use super::*;
-    use crate::executor::Watermark;
 
     #[tokio::test]
     async fn union() {
diff --git a/src/stream/src/executor/watermark_filter.rs b/src/stream/src/executor/watermark_filter.rs
index c966d0411acbb..2aca1251dd058 100644
--- a/src/stream/src/executor/watermark_filter.rs
+++ b/src/stream/src/executor/watermark_filter.rs
@@ -349,8 +349,7 @@ impl<S: StateStore> WatermarkFilterExecutor<S> {
 #[cfg(test)]
 mod tests {
     use itertools::Itertools;
-    use risingwave_common::array::StreamChunk;
-    use risingwave_common::catalog::{ColumnDesc, ColumnId, Field, Schema, TableDesc};
+    use risingwave_common::catalog::{ColumnDesc, ColumnId, Field, TableDesc};
     use risingwave_common::test_prelude::StreamChunkTestExt;
     use risingwave_common::types::Date;
     use risingwave_common::util::epoch::test_epoch;
@@ -359,12 +358,10 @@ mod tests {
     use risingwave_pb::common::ColumnOrder;
     use risingwave_pb::plan_common::PbColumnCatalog;
     use risingwave_storage::memory::MemoryStateStore;
-    use risingwave_storage::table::TableDistribution;
 
     use super::*;
     use crate::executor::test_utils::expr::build_from_pretty;
     use crate::executor::test_utils::{MessageSender, MockSource};
-    use crate::executor::{ActorContext, ExecutorInfo};
 
     const WATERMARK_TYPE: DataType = DataType::Timestamp;
 
diff --git a/src/stream/src/executor/wrapper/epoch_check.rs b/src/stream/src/executor/wrapper/epoch_check.rs
index 0110d8a0d86e6..335c524700f01 100644
--- a/src/stream/src/executor/wrapper/epoch_check.rs
+++ b/src/stream/src/executor/wrapper/epoch_check.rs
@@ -75,7 +75,7 @@ pub async fn epoch_check(info: Arc<ExecutorInfo>, input: impl MessageStream) {
 #[cfg(test)]
 mod tests {
     use assert_matches::assert_matches;
-    use futures::{pin_mut, StreamExt};
+    use futures::pin_mut;
     use risingwave_common::array::StreamChunk;
     use risingwave_common::util::epoch::test_epoch;
 
diff --git a/src/stream/src/task/barrier_manager/tests.rs b/src/stream/src/task/barrier_manager/tests.rs
index 99efb00269b8a..5fbedcdfd0dcc 100644
--- a/src/stream/src/task/barrier_manager/tests.rs
+++ b/src/stream/src/task/barrier_manager/tests.rs
@@ -20,10 +20,8 @@ use std::task::Poll;
 use assert_matches::assert_matches;
 use futures::future::join_all;
 use futures::FutureExt;
-use itertools::Itertools;
 use risingwave_common::util::epoch::test_epoch;
 use risingwave_pb::stream_service::{streaming_control_stream_request, InjectBarrierRequest};
-use tokio::sync::mpsc::unbounded_channel;
 use tokio_stream::wrappers::UnboundedReceiverStream;
 
 use super::*;
diff --git a/src/tests/simulation/tests/integration_tests/scale/plan.rs b/src/tests/simulation/tests/integration_tests/scale/plan.rs
index e1d8148632c62..d39e159fc61d9 100644
--- a/src/tests/simulation/tests/integration_tests/scale/plan.rs
+++ b/src/tests/simulation/tests/integration_tests/scale/plan.rs
@@ -13,7 +13,6 @@
 // limitations under the License.
 
 use std::collections::HashMap;
-use std::default::Default;
 
 use anyhow::Result;
 use itertools::Itertools;