feat(query): Support access Redis data from dictionaries via the dict_get_batch function.

Cargo.toml

@@ -293,6 +293,9 @@ prometheus-client = "0.22"
 prost = { version = "0.12.1" }
 prost-build = { version = "0.12.1" }
 rand = { version = "0.8.5", features = ["small_rng"] }
+redis = { version = "0.27.5", features = [
+    "connection-manager",
+]}
 regex = "1.8.1"
 reqwest = { version = "0.12", default-features = false, features = [
     "json",

src/common/exception/Cargo.toml

@@ -26,6 +26,7 @@ opendal = { workspace = true }
 parquet = { workspace = true }
 paste = { workspace = true }
 prost = { workspace = true }
+redis = { workspace = true }
 reqwest = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }

src/common/exception/src/exception_into.rs

@@ -447,3 +447,9 @@ impl From<sqlx::Error> for ErrorCode {
         ErrorCode::DictionarySourceError(format!("Dictionary Sqlx Error, cause: {}", error))
     }
 }
+
+impl From<redis::RedisError> for ErrorCode {
+    fn from(error: redis::RedisError) -> Self {
+        ErrorCode::DictionarySourceError(format!("Dictionary Redis Error, cause: {}", error))
+    }
+}

src/meta/app/src/schema/dictionary.rs

@@ -116,7 +116,7 @@ impl DictionaryMeta {
             .options
             .get("port")
             .ok_or_else(|| ErrorCode::BadArguments("Miss option `port`"))?;
-        Ok(format!("tcp://{}:{}", host, port))
+        Ok(format!("redis://{}:{}", host, port))
     }
 }
 

src/query/expression/Cargo.toml

@@ -49,6 +49,7 @@ num-bigint = "0.4.6"
 num-traits = "0.2.15"
 rand = { workspace = true }
 recursive = "0.1.1"
+redis = { workspace = true }
 roaring = { version = "0.10.1", features = ["serde"] }
 rust_decimal = "1.26"
 serde = { workspace = true }

src/query/service/Cargo.toml

@@ -153,6 +153,7 @@ poem = { workspace = true }
 prost = { workspace = true }
 rand = { workspace = true }
 recursive = "0.1.1"
+redis = { workspace = true }
 regex = { workspace = true }
 reqwest = { workspace = true }
 rustls = "0.22"

src/query/service/src/pipelines/processors/transforms/transform_async_function.rs

@@ -99,11 +99,10 @@ impl AsyncTransform for TransformAsyncFunction {
                     )
                     .await?;
                 }
-                AsyncFunctionArgument::DictGetFunction(dict_arg) => {
+                AsyncFunctionArgument::DictGetFunction(_) => {
                     self.transform_dict_get(
                         i,
                         &mut data_block,
-                        dict_arg,
                         &async_func_desc.arg_indices,
                         &async_func_desc.data_type,
                     )