fix cmt

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

@@ -15,13 +15,13 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use fluss::client::{FlussConnection, EARLIEST_OFFSET};
+use fluss::client::{EARLIEST_OFFSET, FlussConnection};
 use fluss::config::Config;
 use fluss::error::Result;
 use fluss::metadata::TablePath;
 use fluss::row::InternalRow;
-use std::time::Duration;
 use fluss::rpc::message::OffsetSpec;
+use std::time::Duration;
 use std::time::SystemTime;
 use std::time::UNIX_EPOCH;
 
@@ -37,14 +37,15 @@ pub async fn main() -> Result<()> {
     let conn = FlussConnection::new(config).await?;
     println!("   Connected successfully!");
 
-    let table_path = TablePath::new("fluss".to_owned(), "mahong_log_table_cpp_test_1212".to_owned());
+    let table_path = TablePath::new(
+        "fluss".to_owned(),
+        "mahong_log_table_cpp_test_1212".to_owned(),
+    );
     println!("2) Getting admin...");
     let admin = conn.get_admin().await?;
     println!("   Admin obtained successfully!");
     // Step 1: 通过 admin API 查询时间戳对应的 offset
 
-
-
     println!("3) Getting table: {}", table_path);
     let table = conn.get_table(&table_path).await?;
     println!("   Table obtained successfully!");
@@ -54,16 +55,18 @@ pub async fn main() -> Result<()> {
     println!("   Table has {} buckets", num_buckets);
     let now = SystemTime::now();
     let twenty_minutes_ago = now - Duration::from_secs(20 * 60);
-    
+
     let timestamp_ms = twenty_minutes_ago
         .duration_since(UNIX_EPOCH)
         .expect("Time went backwards")
         .as_secs();
-    let offsets = admin.list_offsets(
-        &table_path, 
-        &(0..num_buckets).collect::<Vec<i32>>(), 
-        OffsetSpec::Timestamp(timestamp_ms as i64)
-    ).await?;
+    let offsets = admin
+        .list_offsets(
+            &table_path,
+            &(0..num_buckets).collect::<Vec<i32>>(),
+            OffsetSpec::Timestamp(timestamp_ms as i64),
+        )
+        .await?;
 
     // println!("4) Creating log scanner...");
     // let log_scanner = table.new_scan().create_log_scanner()?;
@@ -99,9 +102,7 @@ pub async fn main() -> Result<()> {
     // }
 
     println!("\n7) Creating log scanner with projection (columns 0, 1)...");
-    let projected_scanner = table.new_scan()
-        .project(&[0, 2])?
-        .create_log_scanner()?;
+    let projected_scanner = table.new_scan().project(&[0, 2])?.create_log_scanner()?;
     println!("   Projected scanner created successfully!");
 
     println!("8) Subscribing projected scanner to all buckets from EARLIEST_OFFSET...");
@@ -130,7 +131,12 @@ pub async fn main() -> Result<()> {
             }
 
             if i < 10 {
-                println!("  Record {}: id={}, name={}", i, row.get_long(0), row.get_string(1));
+                println!(
+                    "  Record {}: id={}, name={}",
+                    i,
+                    row.get_long(0),
+                    row.get_string(1)
+                );
             }
         }
 
@@ -144,9 +150,5 @@ pub async fn main() -> Result<()> {
             eprintln!("\nColumn pruning verification failed!");
             std::process::exit(1);
         }
-
     }
-
-
-
 }

crates/fluss/src/client/credentials.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()
                 };
@@ -162,10 +166,3 @@ impl Default for CredentialsCache {
         Self::new()
     }
 }
-
-
-
-
-
-
-

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

@@ -184,7 +184,7 @@ impl RemoteLogDownloader {
                 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("://") {
@@ -217,10 +217,15 @@ impl RemoteLogDownloader {
         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);
 
@@ -232,26 +237,37 @@ 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);
+        eprintln!(
+            "[DEBUG] Starting download: {} bytes in {} chunks",
+            file_size, total_chunks
+        );
 
         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!(
+                    "[DEBUG] 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/scanner.rs

@@ -235,7 +235,10 @@ 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());
+        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);
@@ -249,7 +252,10 @@ impl LogFetcher {
             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());
+            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,11 +269,16 @@ 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] 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?;
+                        let s3_props = self
+                            .credentials_cache
+                            .get_or_refresh(&self.conns, &self.metadata)
+                            .await?;
                         eprintln!("[DEBUG] Got {} S3 props", s3_props.len());
                         self.remote_log_downloader.set_s3_props(s3_props);
                         let remote_fetch_info = RemoteLogFetchInfo::from_proto(
@@ -291,8 +302,10 @@ impl LogFetcher {
                                     current_fetch_offset = segment.start_offset;
                                 }
 
-                                eprintln!("[DEBUG] Downloading segment {} from {}", 
-                                    segment.segment_id, remote_fetch_info.remote_log_tablet_dir);
+                                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,
@@ -309,8 +322,10 @@ impl LogFetcher {
                                 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>());
+                                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() {

crates/fluss/src/io/storage_s3.rs

@@ -16,22 +16,22 @@
 // under the License.
 
 use crate::error::Result;
-use opendal::layers::TimeoutLayer;
-use opendal::services::S3Config;
 use opendal::Configurator;
 use opendal::Operator;
+use opendal::layers::TimeoutLayer;
+use opendal::services::S3Config;
 use std::collections::HashMap;
 use std::time::Duration;
 
 pub(crate) fn s3_config_build(props: &HashMap<String, String>) -> Result<Operator> {
     let config = S3Config::from_iter(props.clone())?;
     let op = Operator::from_config(config)?.finish();
-    
+
     // Add timeout layer to prevent hanging on S3 operations
     let timeout_layer = TimeoutLayer::new()
         .with_timeout(Duration::from_secs(10))
         .with_io_timeout(Duration::from_secs(30));
-    
+
     Ok(op.layer(timeout_layer))
 }
 
@@ -46,4 +46,3 @@ pub(crate) fn parse_s3_path(path: &str) -> (&str, &str) {
         None => (path, ""),
     }
 }
-