You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2025/06/13 10:33:11 UTC
(opendal) branch redis-list created (now 11abf9b82)
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a change to branch redis-list
in repository https://gitbox.apache.org/repos/asf/opendal.git
at 11abf9b82 slight refactor
This branch includes the following new commits:
new 0a52169ea feat(services/redis): Add list support
new 11abf9b82 slight refactor
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(opendal) 02/02: slight refactor
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch redis-list
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 11abf9b822500197f8a6609e3f965353def3d2e9
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Fri Jun 13 18:32:55 2025 +0800
slight refactor
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/services/redis/backend.rs | 3 +-
core/src/services/redis/core.rs | 14 +++-----
core/src/services/redis/lister.rs | 72 +++++++++++++++++++++++---------------
3 files changed, 51 insertions(+), 38 deletions(-)
diff --git a/core/src/services/redis/backend.rs b/core/src/services/redis/backend.rs
index 0b3c4b6d0..308e5e858 100644
--- a/core/src/services/redis/backend.rs
+++ b/core/src/services/redis/backend.rs
@@ -374,10 +374,11 @@ impl Access for RedisAccessor {
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
+ // Create lister without unnecessary string allocations
let lister = RedisLister::new(
self.core.clone(),
self.root.clone(),
- path.to_string(),
+ path.to_string(), // We must convert to owned String here since path is borrowed
args.recursive(),
args.limit(),
);
diff --git a/core/src/services/redis/core.rs b/core/src/services/redis/core.rs
index 064c5b994..c51a30ffc 100644
--- a/core/src/services/redis/core.rs
+++ b/core/src/services/redis/core.rs
@@ -210,20 +210,16 @@ impl RedisCore {
let mut conn = self.conn().await?;
let mut cmd = redis::cmd("SCAN");
- cmd.arg(cursor);
- cmd.arg("MATCH");
- cmd.arg(pattern);
+
+ cmd.arg(cursor).arg("MATCH").arg(pattern);
if let Some(count) = count {
- cmd.arg("COUNT");
- cmd.arg(count);
+ cmd.arg("COUNT").arg(count);
}
- let result: (u64, Vec<String>) = cmd
- .query_async(&mut *conn)
+ cmd.query_async(&mut *conn)
.await
- .map_err(format_redis_error)?;
- Ok(result)
+ .map_err(format_redis_error)
}
}
diff --git a/core/src/services/redis/lister.rs b/core/src/services/redis/lister.rs
index 51cd42e27..27f4c3715 100644
--- a/core/src/services/redis/lister.rs
+++ b/core/src/services/redis/lister.rs
@@ -70,48 +70,53 @@ impl RedisLister {
if abs_path.is_empty() {
"*".to_string()
} else {
- let prefix = abs_path.trim_end_matches('/');
- format!("{}/*", prefix)
+ // Normalize path for pattern matching and ensure trailing slash is handled consistently
+ let normalized_path = abs_path.trim_end_matches('/');
+ format!("{}/*", normalized_path)
}
}
fn extract_relative_path(&self, key: &str) -> Option<String> {
let abs_path = build_abs_path(&self.root, &self.path);
- let prefix = if abs_path.is_empty() {
- ""
- } else {
- abs_path.trim_end_matches('/')
- };
- if prefix.is_empty() {
- Some(key.to_string())
- } else if key.starts_with(&format!("{}/", prefix)) {
- Some(key[prefix.len() + 1..].to_string())
+ // For empty path (root listing), all keys are valid
+ if abs_path.is_empty() {
+ return Some(key.to_string());
+ }
+
+ // Normalize the path for consistent prefix matching
+ let prefix = abs_path.trim_end_matches('/');
+ let prefix_with_slash = format!("{}/", prefix);
+
+ // Check if key starts with our path prefix
+ if key.starts_with(&prefix_with_slash) {
+ Some(key[prefix_with_slash.len()..].to_string())
} else {
None
}
}
fn should_include_key(&self, relative_path: &str) -> bool {
+ // For recursive listing, include all keys
if self.recursive {
- true
- } else {
- // For non-recursive, only include direct children
- !relative_path.contains('/')
+ return true;
}
+
+ // For non-recursive, only include direct children (no slashes)
+ !relative_path.contains('/')
}
fn extract_directory_from_key(&self, relative_path: &str) -> Option<String> {
+ // For recursive listing, we never need to extract directories
if self.recursive {
- return None; // Don't emit directories in recursive mode
+ return None;
}
- if let Some(slash_pos) = relative_path.find('/') {
+ // Extract the directory part if there's a slash
+ relative_path.find('/').map(|slash_pos| {
let dir_name = &relative_path[..slash_pos];
- Some(format!("{}/", dir_name))
- } else {
- None
- }
+ format!("{}/", dir_name)
+ })
}
}
@@ -124,16 +129,25 @@ impl oio::PageList for RedisLister {
ctx.token.parse::<u64>().unwrap_or(0)
};
+ // Reuse pattern for all keys in this batch
let pattern = self.build_scan_pattern();
let (next_cursor, keys) = self.core.scan(&pattern, current_cursor, self.limit).await?;
- let mut directories_in_this_batch = HashSet::new();
+ // Only allocate HashSet if needed (for non-recursive listing)
+ let mut directories_in_this_batch = if !self.recursive {
+ Some(HashSet::with_capacity(keys.len() / 2)) // Estimate directory count
+ } else {
+ None
+ };
for key in keys {
if let Some(relative_path) = self.extract_relative_path(&key) {
// Handle directory extraction for non-recursive listing
- if let Some(dir_path) = self.extract_directory_from_key(&relative_path) {
- directories_in_this_batch.insert(dir_path);
+ if let (Some(dir_path), Some(dirs)) = (
+ self.extract_directory_from_key(&relative_path),
+ &mut directories_in_this_batch,
+ ) {
+ dirs.insert(dir_path);
}
// Handle file entries
@@ -146,10 +160,12 @@ impl oio::PageList for RedisLister {
}
}
- // Add directories to entries
- for dir_path in directories_in_this_batch {
- ctx.entries
- .push_back(oio::Entry::new(&dir_path, Metadata::new(EntryMode::DIR)));
+ // Add directories to entries (only for non-recursive mode)
+ if let Some(dirs) = directories_in_this_batch {
+ for dir_path in dirs {
+ ctx.entries
+ .push_back(oio::Entry::new(&dir_path, Metadata::new(EntryMode::DIR)));
+ }
}
// Set next token or mark as done
(opendal) 01/02: feat(services/redis): Add list support
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch redis-list
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 0a52169ea84f8fe3770d5794d11202c8b29663bd
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Fri Jun 13 18:26:41 2025 +0800
feat(services/redis): Add list support
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/services/redis/backend.rs | 18 ++-
core/src/services/redis/core.rs | 25 ++++
core/src/services/redis/lister.rs | 293 +++++++++++++++++++++++++++++++++++++
core/src/services/redis/mod.rs | 2 +
4 files changed, 333 insertions(+), 5 deletions(-)
diff --git a/core/src/services/redis/backend.rs b/core/src/services/redis/backend.rs
index e6817c493..0b3c4b6d0 100644
--- a/core/src/services/redis/backend.rs
+++ b/core/src/services/redis/backend.rs
@@ -31,6 +31,7 @@ use tokio::sync::OnceCell;
use super::core::*;
use super::delete::RedisDeleter;
+use super::lister::RedisLister;
use super::writer::RedisWriter;
use crate::raw::oio;
use crate::raw::*;
@@ -286,6 +287,8 @@ impl RedisAccessor {
write: true,
delete: true,
stat: true,
+ list: true,
+ list_with_recursive: true,
write_can_empty: true,
shared: true,
..Default::default()
@@ -308,7 +311,7 @@ impl RedisAccessor {
impl Access for RedisAccessor {
type Reader = Buffer;
type Writer = RedisWriter;
- type Lister = ();
+ type Lister = oio::PageLister<RedisLister>;
type Deleter = oio::OneShotDeleter<RedisDeleter>;
fn info(&self) -> std::sync::Arc<AccessorInfo> {
@@ -370,10 +373,15 @@ impl Access for RedisAccessor {
))
}
- async fn list(&self, path: &str, _: OpList) -> Result<(RpList, Self::Lister)> {
- let _ = build_abs_path(&self.root, path);
- // Redis doesn't support listing keys, return empty list
- Ok((RpList::default(), ()))
+ async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
+ let lister = RedisLister::new(
+ self.core.clone(),
+ self.root.clone(),
+ path.to_string(),
+ args.recursive(),
+ args.limit(),
+ );
+ Ok((RpList::default(), oio::PageLister::new(lister)))
}
}
diff --git a/core/src/services/redis/core.rs b/core/src/services/redis/core.rs
index 803e99bec..064c5b994 100644
--- a/core/src/services/redis/core.rs
+++ b/core/src/services/redis/core.rs
@@ -200,6 +200,31 @@ impl RedisCore {
let _: () = conn.del(key).await.map_err(format_redis_error)?;
Ok(())
}
+
+ pub async fn scan(
+ &self,
+ pattern: &str,
+ cursor: u64,
+ count: Option<usize>,
+ ) -> Result<(u64, Vec<String>)> {
+ let mut conn = self.conn().await?;
+
+ let mut cmd = redis::cmd("SCAN");
+ cmd.arg(cursor);
+ cmd.arg("MATCH");
+ cmd.arg(pattern);
+
+ if let Some(count) = count {
+ cmd.arg("COUNT");
+ cmd.arg(count);
+ }
+
+ let result: (u64, Vec<String>) = cmd
+ .query_async(&mut *conn)
+ .await
+ .map_err(format_redis_error)?;
+ Ok(result)
+ }
}
pub fn format_redis_error(e: RedisError) -> Error {
diff --git a/core/src/services/redis/lister.rs b/core/src/services/redis/lister.rs
new file mode 100644
index 000000000..51cd42e27
--- /dev/null
+++ b/core/src/services/redis/lister.rs
@@ -0,0 +1,293 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::collections::HashSet;
+use std::sync::Arc;
+
+use super::core::RedisCore;
+use crate::raw::oio;
+use crate::raw::*;
+use crate::*;
+
+/// Redis lister that uses SCAN cursor for efficient key discovery
+///
+/// This lister leverages Redis's SCAN command to iterate through keys efficiently
+/// without blocking the server. It supports:
+/// - Cursor-based pagination for large key sets
+/// - Hierarchical directory simulation using "/" delimiter
+/// - Configurable scan batch size via OpList::limit()
+/// - Both recursive and non-recursive listing modes
+pub struct RedisLister {
+ core: Arc<RedisCore>,
+ root: String,
+ path: String,
+ recursive: bool,
+ /// Limit for SCAN command batch size, taken from OpList::limit()
+ limit: Option<usize>,
+}
+
+impl RedisLister {
+ /// Create a new Redis lister
+ ///
+ /// # Arguments
+ /// * `core` - Redis core instance for executing commands
+ /// * `root` - Root path for the accessor
+ /// * `path` - Target path to list
+ /// * `recursive` - Whether to list recursively
+ /// * `limit` - Batch size for SCAN operations (from OpList::limit())
+ pub fn new(
+ core: Arc<RedisCore>,
+ root: String,
+ path: String,
+ recursive: bool,
+ limit: Option<usize>,
+ ) -> Self {
+ Self {
+ core,
+ root,
+ path,
+ recursive,
+ limit,
+ }
+ }
+
+ fn build_scan_pattern(&self) -> String {
+ let abs_path = build_abs_path(&self.root, &self.path);
+ if abs_path.is_empty() {
+ "*".to_string()
+ } else {
+ let prefix = abs_path.trim_end_matches('/');
+ format!("{}/*", prefix)
+ }
+ }
+
+ fn extract_relative_path(&self, key: &str) -> Option<String> {
+ let abs_path = build_abs_path(&self.root, &self.path);
+ let prefix = if abs_path.is_empty() {
+ ""
+ } else {
+ abs_path.trim_end_matches('/')
+ };
+
+ if prefix.is_empty() {
+ Some(key.to_string())
+ } else if key.starts_with(&format!("{}/", prefix)) {
+ Some(key[prefix.len() + 1..].to_string())
+ } else {
+ None
+ }
+ }
+
+ fn should_include_key(&self, relative_path: &str) -> bool {
+ if self.recursive {
+ true
+ } else {
+ // For non-recursive, only include direct children
+ !relative_path.contains('/')
+ }
+ }
+
+ fn extract_directory_from_key(&self, relative_path: &str) -> Option<String> {
+ if self.recursive {
+ return None; // Don't emit directories in recursive mode
+ }
+
+ if let Some(slash_pos) = relative_path.find('/') {
+ let dir_name = &relative_path[..slash_pos];
+ Some(format!("{}/", dir_name))
+ } else {
+ None
+ }
+ }
+}
+
+impl oio::PageList for RedisLister {
+ async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
+ // Parse cursor from token
+ let current_cursor = if ctx.token.is_empty() {
+ 0
+ } else {
+ ctx.token.parse::<u64>().unwrap_or(0)
+ };
+
+ let pattern = self.build_scan_pattern();
+ let (next_cursor, keys) = self.core.scan(&pattern, current_cursor, self.limit).await?;
+
+ let mut directories_in_this_batch = HashSet::new();
+
+ for key in keys {
+ if let Some(relative_path) = self.extract_relative_path(&key) {
+ // Handle directory extraction for non-recursive listing
+ if let Some(dir_path) = self.extract_directory_from_key(&relative_path) {
+ directories_in_this_batch.insert(dir_path);
+ }
+
+ // Handle file entries
+ if self.should_include_key(&relative_path) {
+ ctx.entries.push_back(oio::Entry::new(
+ &relative_path,
+ Metadata::new(EntryMode::FILE),
+ ));
+ }
+ }
+ }
+
+ // Add directories to entries
+ for dir_path in directories_in_this_batch {
+ ctx.entries
+ .push_back(oio::Entry::new(&dir_path, Metadata::new(EntryMode::DIR)));
+ }
+
+ // Set next token or mark as done
+ if next_cursor == 0 {
+ ctx.done = true;
+ } else {
+ ctx.token = next_cursor.to_string();
+ }
+
+ Ok(())
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use std::time::Duration;
+ use tokio::sync::OnceCell;
+
+ fn create_test_core() -> Arc<RedisCore> {
+ Arc::new(RedisCore {
+ addr: "redis://127.0.0.1:6379".to_string(),
+ client: None,
+ cluster_client: None,
+ conn: OnceCell::new(),
+ default_ttl: Some(Duration::from_secs(60)),
+ })
+ }
+
+ #[test]
+ fn test_build_scan_pattern() {
+ let core = create_test_core();
+
+ // Test root path
+ let lister = RedisLister::new(core.clone(), "/".to_string(), "".to_string(), false, None);
+ assert_eq!(lister.build_scan_pattern(), "*");
+
+ // Test subdirectory path
+ let lister = RedisLister::new(
+ core.clone(),
+ "/".to_string(),
+ "test/dir".to_string(),
+ false,
+ None,
+ );
+ assert_eq!(lister.build_scan_pattern(), "test/dir/*");
+
+ // Test with custom root
+ let lister = RedisLister::new(
+ core,
+ "/myroot/".to_string(),
+ "subdir".to_string(),
+ true,
+ None,
+ );
+ assert_eq!(lister.build_scan_pattern(), "myroot/subdir/*");
+ }
+
+ #[test]
+ fn test_extract_relative_path() {
+ let core = create_test_core();
+
+ // Test with root path
+ let lister = RedisLister::new(core.clone(), "/".to_string(), "".to_string(), false, None);
+ assert_eq!(
+ lister.extract_relative_path("file.txt"),
+ Some("file.txt".to_string())
+ );
+ assert_eq!(
+ lister.extract_relative_path("dir/file.txt"),
+ Some("dir/file.txt".to_string())
+ );
+
+ // Test with subdirectory
+ let lister = RedisLister::new(core, "/".to_string(), "test".to_string(), false, None);
+ assert_eq!(
+ lister.extract_relative_path("test/file.txt"),
+ Some("file.txt".to_string())
+ );
+ assert_eq!(
+ lister.extract_relative_path("test/dir/file.txt"),
+ Some("dir/file.txt".to_string())
+ );
+ assert_eq!(lister.extract_relative_path("other/file.txt"), None);
+ }
+
+ #[test]
+ fn test_should_include_key() {
+ let core = create_test_core();
+
+ // Test recursive mode
+ let lister = RedisLister::new(core.clone(), "/".to_string(), "".to_string(), true, None);
+ assert!(lister.should_include_key("file.txt"));
+ assert!(lister.should_include_key("dir/file.txt"));
+ assert!(lister.should_include_key("deep/nested/file.txt"));
+
+ // Test non-recursive mode
+ let lister = RedisLister::new(core, "/".to_string(), "".to_string(), false, None);
+ assert!(lister.should_include_key("file.txt"));
+ assert!(!lister.should_include_key("dir/file.txt"));
+ assert!(!lister.should_include_key("deep/nested/file.txt"));
+ }
+
+ #[test]
+ fn test_extract_directory_from_key() {
+ let core = create_test_core();
+
+ // Test recursive mode - should not extract directories
+ let lister = RedisLister::new(core.clone(), "/".to_string(), "".to_string(), true, None);
+ assert_eq!(lister.extract_directory_from_key("dir/file.txt"), None);
+
+ // Test non-recursive mode - should extract directories
+ let lister = RedisLister::new(core, "/".to_string(), "".to_string(), false, None);
+ assert_eq!(
+ lister.extract_directory_from_key("dir/file.txt"),
+ Some("dir/".to_string())
+ );
+ assert_eq!(lister.extract_directory_from_key("file.txt"), None);
+ assert_eq!(
+ lister.extract_directory_from_key("deep/nested/file.txt"),
+ Some("deep/".to_string())
+ );
+ }
+
+ #[test]
+ fn test_redis_lister_creation() {
+ let core = create_test_core();
+ let lister = RedisLister::new(
+ core,
+ "/test/".to_string(),
+ "subdir".to_string(),
+ false,
+ Some(100),
+ );
+
+ assert_eq!(lister.root, "/test/");
+ assert_eq!(lister.path, "subdir");
+ assert!(!lister.recursive);
+ assert_eq!(lister.limit, Some(100));
+ }
+}
diff --git a/core/src/services/redis/mod.rs b/core/src/services/redis/mod.rs
index 3d8975fae..3f1bb6191 100644
--- a/core/src/services/redis/mod.rs
+++ b/core/src/services/redis/mod.rs
@@ -22,6 +22,8 @@ mod core;
#[cfg(feature = "services-redis")]
mod delete;
#[cfg(feature = "services-redis")]
+mod lister;
+#[cfg(feature = "services-redis")]
mod writer;
#[cfg(feature = "services-redis")]