address comments

Author: 赵海源

crates/examples/Cargo.toml

@@ -32,28 +32,4 @@ serde_json = { workspace = true }
 opendal = { workspace = true }
 [[example]]
 name = "example-table"
-path = "src/example_table.rs"
-
-[[example]]
-name = "scan-log-table"
-path = "src/scan_log_table.rs"
-
-[[example]]
-name = "scan-from-earliest"
-path = "src/scan_from_earliest.rs"
-
-[[example]]
-name = "scan-from-ts"
-path = "src/scan_from_ts.rs"
-
-[[example]]
-name = "scan-local-only"
-path = "src/scan_local_only.rs"
-
-[[example]]
-name = "debug-credentials"
-path = "src/debug_credentials.rs"
-
-[[example]]
-name = "debug-s3-download"
-path = "src/debug_s3_download.rs"
+path = "src/example_table.rs"

crates/examples/src/scan_from_ts.rs

@@ -17,8 +17,8 @@
 
 use crate::client::metadata::Metadata;
 use crate::error::{Error, Result};
-use crate::rpc::message::GetSecurityTokenRequest;
 use crate::rpc::RpcClient;
+use crate::rpc::message::GetSecurityTokenRequest;
 use parking_lot::RwLock;
 use serde::Deserialize;
 use std::collections::HashMap;
@@ -60,7 +60,11 @@ impl CachedToken {
             if let Some((opendal_key, transform)) = convert_hadoop_key_to_opendal(key) {
                 let final_value = if transform {
                     // Invert boolean value (path_style_access -> enable_virtual_host_style)
-                    if value == "true" { "false".to_string() } else { "true".to_string() }
+                    if value == "true" {
+                        "false".to_string()
+                    } else {
+                        "true".to_string()
+                    }
                 } else {
                     value.clone()
                 };
@@ -76,17 +80,10 @@ impl CachedToken {
 /// needs_inversion is true for path_style_access -> enable_virtual_host_style conversion
 fn convert_hadoop_key_to_opendal(hadoop_key: &str) -> Option<(String, bool)> {
     match hadoop_key {
-        // Standard S3A keys
         "fs.s3a.endpoint" => Some(("endpoint".to_string(), false)),
         "fs.s3a.endpoint.region" => Some(("region".to_string(), false)),
-        // path.style.access = false means virtual_host_style = true (inverted)
         "fs.s3a.path.style.access" => Some(("enable_virtual_host_style".to_string(), true)),
         "fs.s3a.connection.ssl.enabled" => None,
-        // Red-S3 keys (Fluss custom format)
-        "fs.red-s3.endpoint" => Some(("endpoint".to_string(), false)),
-        "fs.red-s3.region" => Some(("region".to_string(), false)),
-        "fs.red-s3.path-style-access" => Some(("enable_virtual_host_style".to_string(), true)),
-        "fs.red-s3.connection.ssl.enabled" => None,
         _ => None,
     }
 }
@@ -162,10 +159,3 @@ impl Default for CredentialsCache {
         Self::new()
     }
 }
-
-
-
-
-
-
-

crates/fluss/src/client/credentials.rs

@@ -175,16 +175,6 @@ impl RemoteLogDownloader {
         local_path: &Path,
         s3_props: &HashMap<String, String>,
     ) -> Result<PathBuf> {
-        eprintln!("[DEBUG] download_file called: remote_path={}", remote_path);
-        eprintln!("[DEBUG] s3_props count: {}", s3_props.len());
-        for (k, v) in s3_props {
-            if k.contains("key") || k.contains("secret") {
-                eprintln!("[DEBUG]   {} = {}...", k, &v[..std::cmp::min(8, v.len())]);
-            } else {
-                eprintln!("[DEBUG]   {} = {}", k, v);
-            }
-        }
-        
         // Handle both URL (e.g., "s3://bucket/path") and local file paths
         // If the path doesn't contain "://", treat it as a local file path
         let remote_log_tablet_dir_url = if remote_log_tablet_dir.contains("://") {
@@ -212,17 +202,20 @@ impl RemoteLogDownloader {
         // Timeout for remote storage operations (30 seconds)
         const REMOTE_OP_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
 
-        eprintln!("[DEBUG] Calling stat on: {}", relative_path);
         // Get file metadata to know the size with timeout
         let stat_future = op.stat(relative_path);
         let meta = tokio::time::timeout(REMOTE_OP_TIMEOUT, stat_future)
             .await
-            .map_err(|_| Error::Io(io::Error::new(
-                io::ErrorKind::TimedOut,
-                format!("Timeout getting file metadata from remote storage: {}", remote_path)
-            )))??;
+            .map_err(|_| {
+                Error::Io(io::Error::new(
+                    io::ErrorKind::TimedOut,
+                    format!(
+                        "Timeout getting file metadata from remote storage: {}",
+                        remote_path
+                    ),
+                ))
+            })??;
         let file_size = meta.content_length();
-        eprintln!("[DEBUG] stat succeeded, file_size={}", file_size);
 
         // Create local file for writing
         let mut local_file = tokio::fs::File::create(local_path).await?;
@@ -232,26 +225,33 @@ impl RemoteLogDownloader {
         const CHUNK_SIZE: u64 = 8 * 1024 * 1024; // 8MB chunks for efficient streaming
         let mut offset = 0u64;
         let mut chunk_count = 0u64;
-        let total_chunks = (file_size + CHUNK_SIZE - 1) / CHUNK_SIZE;
-        eprintln!("[DEBUG] Starting download: {} bytes in {} chunks", file_size, total_chunks);
+        let total_chunks = file_size.div_ceil(CHUNK_SIZE);
 
         while offset < file_size {
             let end = std::cmp::min(offset + CHUNK_SIZE, file_size);
             let range = offset..end;
             chunk_count += 1;
 
             if chunk_count <= 3 || chunk_count % 10 == 0 {
-                eprintln!("[DEBUG] Reading chunk {}/{} (offset {})", chunk_count, total_chunks, offset);
+                eprintln!(
+                    "Remote log download: reading chunk {}/{} (offset {})",
+                    chunk_count, total_chunks, offset
+                );
             }
 
             // Read chunk from remote storage with timeout
             let read_future = op.read_with(relative_path).range(range.clone());
             let chunk = tokio::time::timeout(REMOTE_OP_TIMEOUT, read_future)
                 .await
-                .map_err(|_| Error::Io(io::Error::new(
-                    io::ErrorKind::TimedOut,
-                    format!("Timeout reading chunk from remote storage: {} at offset {}", remote_path, offset)
-                )))??;
+                .map_err(|_| {
+                    Error::Io(io::Error::new(
+                        io::ErrorKind::TimedOut,
+                        format!(
+                            "Timeout reading chunk from remote storage: {} at offset {}",
+                            remote_path, offset
+                        ),
+                    ))
+                })??;
             let bytes = chunk.to_bytes();
 
             // Write chunk to local file

crates/fluss/src/client/table/remote_log.rs

@@ -235,21 +235,17 @@ impl LogFetcher {
 
     async fn send_fetches_and_collect(&self) -> Result<HashMap<TableBucket, Vec<ScanRecord>>> {
         let fetch_request = self.prepare_fetch_log_requests().await;
-        eprintln!("[DEBUG] prepare_fetch_log_requests returned {} requests", fetch_request.len());
         let mut result: HashMap<TableBucket, Vec<ScanRecord>> = HashMap::new();
         for (leader, fetch_request) in fetch_request {
-            eprintln!("[DEBUG] Sending fetch request to leader {}", leader);
             let cluster = self.metadata.get_cluster();
             let server_node = cluster
                 .get_tablet_server(leader)
                 .expect("todo: handle leader not exist.");
             let con = self.conns.get_connection(server_node).await?;
 
-            eprintln!("[DEBUG] Sending RPC request...");
             let fetch_response = con
                 .request(crate::rpc::message::FetchLogRequest::new(fetch_request))
                 .await?;
-            eprintln!("[DEBUG] Got RPC response with {} tables", fetch_response.tables_resp.len());
 
             for pb_fetch_log_resp in fetch_response.tables_resp {
                 let table_id = pb_fetch_log_resp.table_id;
@@ -263,12 +259,10 @@ impl LogFetcher {
                     if let Some(ref remote_log_fetch_info) =
                         fetch_log_for_bucket.remote_log_fetch_info
                     {
-                        eprintln!("[DEBUG] Bucket {} requires remote log fetch with {} segments", 
-                            bucket, remote_log_fetch_info.remote_log_segments.len());
-                        eprintln!("[DEBUG] Getting S3 credentials...");
-                        let s3_props =
-                            self.credentials_cache.get_or_refresh(&self.conns, &self.metadata).await?;
-                        eprintln!("[DEBUG] Got {} S3 props", s3_props.len());
+                        let s3_props = self
+                            .credentials_cache
+                            .get_or_refresh(&self.conns, &self.metadata)
+                            .await?;
                         self.remote_log_downloader.set_s3_props(s3_props);
                         let remote_fetch_info = RemoteLogFetchInfo::from_proto(
                             remote_log_fetch_info,
@@ -291,8 +285,6 @@ impl LogFetcher {
                                     current_fetch_offset = segment.start_offset;
                                 }
 
-                                eprintln!("[DEBUG] Downloading segment {} from {}", 
-                                    segment.segment_id, remote_fetch_info.remote_log_tablet_dir);
                                 let download_future =
                                     self.remote_log_downloader.request_remote_log(
                                         &remote_fetch_info.remote_log_tablet_dir,
@@ -306,11 +298,8 @@ impl LogFetcher {
                                     high_watermark,
                                     self.read_context.clone(),
                                 );
-                                eprintln!("[DEBUG] Waiting for download to complete...");
                                 let remote_records =
                                     pending_fetch.convert_to_completed_fetch().await?;
-                                eprintln!("[DEBUG] Download completed, got {} records", 
-                                    remote_records.values().map(|v| v.len()).sum::<usize>());
                                 // Update offset and merge results
                                 for (tb, records) in remote_records {
                                     if let Some(last_record) = records.last() {