diff --git a/Cargo.lock b/Cargo.lock
index 102086916e78d..a235d907b321c 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -10700,123 +10700,18 @@ version = "1.9.0-alpha"
 dependencies = [
  "anyhow",
  "apache-avro 0.16.0",
- "arrow-array 50.0.0",
- "arrow-row 50.0.0",
- "arrow-schema 50.0.0",
- "arrow-select 50.0.0",
- "assert_matches",
- "async-nats",
- "async-trait",
- "auto_enums",
- "auto_impl",
- "await-tree",
- "aws-config",
- "aws-credential-types",
- "aws-msk-iam-sasl-signer",
- "aws-sdk-kinesis",
- "aws-sdk-s3",
- "aws-smithy-http",
- "aws-smithy-runtime-api",
- "aws-smithy-types",
- "aws-smithy-types-convert",
- "aws-types",
- "base64 0.22.0",
- "byteorder",
- "bytes",
- "cfg-or-panic",
  "chrono",
- "clickhouse",
- "criterion",
- "csv",
- "deltalake",
- "duration-str",
- "easy-ext",
- "enum-as-inner 0.6.0",
- "expect-test",
- "futures",
- "futures-async-stream",
- "gcp-bigquery-client",
- "glob",
- "google-cloud-bigquery",
- "google-cloud-gax",
- "google-cloud-googleapis",
- "google-cloud-pubsub",
- "http 0.2.9",
- "icelake",
- "indexmap 1.9.3",
  "itertools 0.12.1",
- "jni",
  "jsonbb",
- "jsonschema-transpiler",
- "jsonwebtoken",
- "madsim-rdkafka",
- "madsim-tokio",
- "madsim-tonic",
- "maplit",
- "moka",
- "mysql_async",
- "mysql_common",
- "nexmark",
  "num-bigint",
- "opendal",
- "openssl",
- "parking_lot 0.12.1",
- "paste",
- "pg_bigdecimal",
- "postgres-openssl",
- "pretty_assertions",
- "prometheus",
- "prost 0.12.1",
- "prost-build 0.12.1",
- "prost-reflect",
- "prost-types 0.12.1",
- "protobuf-native",
- "protobuf-src",
- "pulsar",
- "quote",
- "rand",
- "redis",
- "regex",
- "reqwest 0.12.4",
  "risingwave_common",
- "risingwave_common_estimate_size",
- "risingwave_jni_core",
- "risingwave_object_store",
  "risingwave_pb",
- "risingwave_rpc_client",
- "rumqttc",
  "rust_decimal",
- "rustls-native-certs 0.7.0",
- "rustls-pemfile 2.1.1",
- "rustls-pki-types",
- "rw_futures_util",
- "serde",
- "serde_derive",
- "serde_json",
- "serde_with",
- "serde_yaml",
- "simd-json",
- "strum 0.26.1",
- "strum_macros 0.26.1",
- "syn 1.0.109",
- "tempfile",
  "thiserror",
  "thiserror-ext",
  "time",
- "tokio-postgres",
- "tokio-retry",
- "tokio-stream",
- "tokio-util",
  "tracing",
- "tracing-subscriber",
- "tracing-test",
- "url",
- "urlencoding",
- "uuid",
- "walkdir",
- "with_options",
  "workspace-hack",
- "yup-oauth2",
 ]
 
 [[package]]
diff --git a/src/connector/encdec/Cargo.toml b/src/connector/encdec/Cargo.toml
index 664bc84791cdc..ab25981ca7f69 100644
--- a/src/connector/encdec/Cargo.toml
+++ b/src/connector/encdec/Cargo.toml
@@ -22,162 +22,23 @@ apache-avro = { git = "https://github.com/risingwavelabs/avro", rev = "5349b0c7b
     "bzip",
     "xz",
 ] }
-arrow-array = { workspace = true }
-arrow-row = { workspace = true }
-arrow-schema = { workspace = true }
-arrow-select = { workspace = true }
-assert_matches = "1"
-async-nats = "0.35"
-async-trait = "0.1"
-auto_enums = { workspace = true }
-auto_impl = "1"
-await-tree = { workspace = true }
-aws-config = { workspace = true }
-aws-credential-types = { workspace = true }
-aws-msk-iam-sasl-signer = "1.0.0"
-aws-sdk-kinesis = { workspace = true }
-aws-sdk-s3 = { workspace = true }
-aws-smithy-http = { workspace = true }
-aws-smithy-runtime-api = { workspace = true }
-aws-smithy-types = { workspace = true }
-aws-smithy-types-convert = { version = "0.60.1", features = ["convert-chrono"] }
-aws-types = { workspace = true }
-base64 = "0.22"
-byteorder = "1"
-bytes = { version = "1", features = ["serde"] }
-cfg-or-panic = "0.2"
 chrono = { version = "0.4", default-features = false, features = [
     "clock",
     "std",
 ] }
-clickhouse = { git = "https://github.com/risingwavelabs/clickhouse.rs", rev = "d38c8b6391af098b724c114e5a4746aedab6ab8e", features = [
-    "time",
-] }
-csv = "1.3"
-deltalake = { workspace = true }
-duration-str = "0.10.0"
-easy-ext = "1"
-enum-as-inner = "0.6"
-futures = { version = "0.3", default-features = false, features = ["alloc"] }
-futures-async-stream = { workspace = true }
-gcp-bigquery-client = "0.18.0"
-glob = "0.3"
-google-cloud-bigquery = { version = "0.9.0", features = ["auth"] }
-google-cloud-gax = "0.17.0"
-google-cloud-googleapis = { version = "0.13", features = ["pubsub", "bigquery"] }
-google-cloud-pubsub = "0.25"
-http = "0.2"
-icelake = { workspace = true }
-indexmap = { version = "1.9.3", features = ["serde"] }
 itertools = { workspace = true }
-jni = { version = "0.21.1", features = ["invocation"] }
 jsonbb = { workspace = true }
-jsonwebtoken = "9.2.0"
-jst = { package = 'jsonschema-transpiler', git = "https://github.com/mozilla/jsonschema-transpiler", rev = "c1a89d720d118843d8bcca51084deb0ed223e4b4" }
-maplit = "1.0.2"
-moka = { version = "0.12", features = ["future"] }
-mysql_async = { version = "0.34", default-features = false, features = [
-    "default",
-] }
-mysql_common = { version = "0.32", default-features = false, features = [
-    "chrono",
-] }
-nexmark = { version = "0.2", features = ["serde"] }
 num-bigint = "0.4"
-opendal = "0.45"
-openssl = "0.10"
-parking_lot = { workspace = true }
-paste = "1"
-pg_bigdecimal = { git = "https://github.com/risingwavelabs/rust-pg_bigdecimal", rev = "0b7893d88894ca082b4525f94f812da034486f7c" }
-postgres-openssl = "0.5.0"
-prometheus = { version = "0.13", features = ["process"] }
-prost = { version = "0.12", features = ["no-recursion-limit"] }
-prost-reflect = "0.13"
-prost-types = "0.12"
-protobuf-native = "0.2.2"
-pulsar = { version = "6.3", default-features = false, features = [
-    "tokio-runtime",
-    "telemetry",
-    "auth-oauth2",
-    "lz4",
-    "zstd",
-] }
-rdkafka = { workspace = true, features = [
-    "cmake-build",
-    "ssl",
-    "gssapi",
-    "zstd",
-] }
-redis = { version = "0.25", features = ["aio", "tokio-comp", "async-std-comp","cluster-async"] }
-regex = "1.4"
-reqwest = { version = "0.12.2", features = ["json", "stream"] }
 risingwave_common = { workspace = true }
-risingwave_common_estimate_size = { workspace = true }
-risingwave_jni_core = { workspace = true }
-risingwave_object_store = { workspace = true }
 risingwave_pb = { workspace = true }
-risingwave_rpc_client = { workspace = true }
-rumqttc = { version = "0.24.0", features = ["url"] }
 rust_decimal = "1"
-rustls-native-certs = "0.7"
-rustls-pemfile = "2"
-rustls-pki-types = "1"
-rw_futures_util = { workspace = true }
-serde = { version = "1", features = ["derive", "rc"] }
-serde_derive = "1"
-serde_json = "1"
-serde_with = { version = "3", features = ["json"] }
-simd-json = "0.13.3"
-strum = "0.26"
-strum_macros = "0.26"
-tempfile = "3"
 thiserror = "1"
 thiserror-ext = { workspace = true }
 time = "0.3.30"
-tokio = { version = "0.2", package = "madsim-tokio", features = [
-    "rt",
-    "rt-multi-thread",
-    "sync",
-    "macros",
-    "time",
-    "signal",
-    "fs",
-] }
-tokio-postgres = { version = "0.7", features = ["with-uuid-1"] }
-tokio-retry = "0.3"
-tokio-stream = "0.1"
-tokio-util = { version = "0.7", features = ["codec", "io"] }
-tonic = { workspace = true }
 tracing = "0.1"
-url = "2"
-urlencoding = "2"
-uuid = { version = "1", features = ["v4", "fast-rng"] }
-with_options = { path = "../with_options" }
-yup-oauth2 = "8.3"
 
 [target.'cfg(not(madsim))'.dependencies]
 workspace-hack = { path = "../../workspace-hack" }
 
-[dev-dependencies]
-assert_matches = "1"
-criterion = { workspace = true, features = ["async_tokio", "async"] }
-deltalake = { workspace = true, features = ["datafusion"] }
-expect-test = "1"
-paste = "1"
-pretty_assertions = "1"
-quote = "1"
-rand = { workspace = true }
-serde = { version = "1", features = ["derive"] }
-serde_yaml = "0.9"
-syn = { version = "1", features = ["full"] }
-tempfile = "3"
-tracing-subscriber = "0.3"
-tracing-test = "0.2"
-walkdir = "2"
-
-[build-dependencies]
-prost-build = "0.12"
-protobuf-src = "1"
-
 [lints]
 workspace = true
diff --git a/src/connector/src/parser/avro/parser.rs b/src/connector/src/parser/avro/parser.rs
index 86c70cd0ddfc7..9cc0090be1e20 100644
--- a/src/connector/src/parser/avro/parser.rs
+++ b/src/connector/src/parser/avro/parser.rs
@@ -198,6 +198,7 @@ mod test {
     use risingwave_common::catalog::ColumnId;
     use risingwave_common::row::Row;
     use risingwave_common::types::{DataType, Date};
+    use risingwave_common::util::iter_util::ZipEqFast;
     use risingwave_pb::catalog::StreamSourceInfo;
     use risingwave_pb::plan_common::{PbEncodeType, PbFormatType};
     use url::Url;
@@ -332,7 +333,7 @@ mod test {
             record
                 .fields
                 .iter()
-                .zip_eq(row.iter())
+                .zip_eq_fast(row.iter())
                 .format_with("\n", |(avro, datum), f| {
                     f(&format_args!("{:?} => {:?}", avro, datum))
                 })