feat: support read s3

Author: 赵海源

Cargo.toml

@@ -36,3 +36,8 @@ tokio = { version = "1.44.2", features = ["full"] }
 clap = { version = "4.5.37", features = ["derive"] }
 arrow = { version = "57.0.0", features = ["ipc_compression"] }
 chrono = { version = "0.4", features = ["clock", "std", "wasmbind"] }
+
+serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0"
+opendal = { version = "0.53", features = ["services-s3"] }
+

crates/examples/Cargo.toml

@@ -26,7 +26,34 @@ version = { workspace = true }
 [dependencies]
 fluss = { workspace = true }
 tokio = { workspace = true }
-clap = { workspace = true}
+clap = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+opendal = { workspace = true }
 [[example]]
 name = "example-table"
-path = "src/example_table.rs"
+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"

crates/examples/src/scan_from_ts.rs

@@ -0,0 +1,152 @@
+// 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 fluss::client::{FlussConnection, EARLIEST_OFFSET};
+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::SystemTime;
+use std::time::UNIX_EPOCH;
+
+#[tokio::main]
+pub async fn main() -> Result<()> {
+    println!("Starting Rust scan from earliest example...");
+    println!("Using EARLIEST_OFFSET = {}", EARLIEST_OFFSET);
+
+    let mut config = Config::default();
+    config.bootstrap_server = Some("10.147.136.86:9123".to_string());
+
+    println!("1) Connecting to Fluss...");
+    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());
+    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!");
+
+    let table_info = table.table_info();
+    let num_buckets = table_info.num_buckets;
+    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?;
+
+    // println!("4) Creating log scanner...");
+    // let log_scanner = table.new_scan().create_log_scanner()?;
+    // println!("   Log scanner created successfully!");
+
+    // println!("5) Subscribing to all buckets from timestamp {}...", timestamp_ms);
+    // for (bucket_id, offset) in offsets {
+    //     log_scanner.subscribe(bucket_id, offset).await?;
+    //     println!("   Subscribed to bucket {} from timestamp {} offset {}", bucket_id, timestamp_ms, offset);
+    // }
+
+    // println!("6) Polling records (timeout: 10 seconds)...");
+    // let scan_records = log_scanner.poll(Duration::from_secs(10)).await?;
+
+    // let record_count = scan_records.count();
+    // println!("Scanned records: {}", record_count);
+    // for (i, record) in scan_records.into_iter().enumerate() {
+    //     let row = record.row();
+    //     if i < 10 {
+    //         println!(
+    //             " offset={} id={} number={} value={} ts={}",
+    //             record.offset(),
+    //             row.get_long(0),
+    //             row.get_int(1),
+    //             row.get_string(2),
+    //             record.timestamp()
+    //         );
+    //     }
+    // }
+
+    // if record_count > 10 {
+    //     println!(" ... and {} more records", record_count - 10);
+    // }
+
+    println!("\n7) Creating log scanner with projection (columns 0, 1)...");
+    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...");
+    for bucket_id in 0..num_buckets {
+        // projected_scanner.subscribe(bucket_id, EARLIEST_OFFSET).await?;
+        projected_scanner.subscribe(bucket_id, 278807821).await?;
+        println!("   Subscribed to bucket {} from earliest", bucket_id);
+    }
+
+    loop {
+        println!("9) Polling projected records (timeout: 10 seconds)...");
+        let projected_records = projected_scanner.poll(Duration::from_secs(10)).await?;
+
+        let projected_count = projected_records.count();
+        println!("Projected records: {}", projected_count);
+
+        let mut projection_verified = true;
+        for (i, record) in projected_records.into_iter().enumerate() {
+            let row = record.row();
+            let field_count = row.get_field_count();
+
+            if field_count != 2 {
+                eprintln!("ERROR: Record {} has {} fields, expected 2", i, field_count);
+                projection_verified = false;
+                continue;
+            }
+
+            if i < 10 {
+                println!("  Record {}: id={}, name={}", i, row.get_long(0), row.get_string(1));
+            }
+        }
+
+        if projected_count > 10 {
+            println!("  ... and {} more records", projected_count - 10);
+        }
+
+        if projection_verified {
+            println!("\nColumn pruning verification passed!");
+        } else {
+            eprintln!("\nColumn pruning verification failed!");
+            std::process::exit(1);
+        }
+
+    }
+
+
+
+}

crates/fluss/Cargo.toml

@@ -23,11 +23,12 @@ name = "fluss"
 build = "src/build.rs"
 
 [features]
-default = ["storage-memory", "storage-fs"]
-storage-all = ["storage-memory", "storage-fs"]
+default = ["storage-memory", "storage-fs", "storage-s3"]
+storage-all = ["storage-memory", "storage-fs", "storage-s3"]
 
 storage-memory = ["opendal/services-memory"]
 storage-fs = ["opendal/services-fs"]
+storage-s3 = ["opendal/services-s3"]
 integration_tests = []
 
 [dependencies]

crates/fluss/src/client/credentials.rs

@@ -0,0 +1,171 @@
+// 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 crate::client::metadata::Metadata;
+use crate::error::{Error, Result};
+use crate::rpc::message::GetSecurityTokenRequest;
+use crate::rpc::RpcClient;
+use parking_lot::RwLock;
+use serde::Deserialize;
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+const CACHE_TTL: Duration = Duration::from_secs(3600);
+
+#[derive(Debug, Deserialize)]
+struct Credentials {
+    access_key_id: String,
+    access_key_secret: String,
+    security_token: Option<String>,
+}
+
+struct CachedToken {
+    access_key_id: String,
+    secret_access_key: String,
+    security_token: Option<String>,
+    addition_infos: HashMap<String, String>,
+    cached_at: Instant,
+}
+
+impl CachedToken {
+    fn to_s3_props(&self) -> HashMap<String, String> {
+        let mut props = HashMap::new();
+
+        props.insert("access_key_id".to_string(), self.access_key_id.clone());
+        props.insert(
+            "secret_access_key".to_string(),
+            self.secret_access_key.clone(),
+        );
+
+        if let Some(token) = &self.security_token {
+            props.insert("security_token".to_string(), token.clone());
+        }
+
+        for (key, value) in &self.addition_infos {
+            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() }
+                } else {
+                    value.clone()
+                };
+                props.insert(opendal_key, final_value);
+            }
+        }
+
+        props
+    }
+}
+
+/// Returns (opendal_key, needs_inversion)
+/// 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,
+    }
+}
+
+pub struct CredentialsCache {
+    inner: RwLock<Option<CachedToken>>,
+}
+
+impl CredentialsCache {
+    pub fn new() -> Self {
+        Self {
+            inner: RwLock::new(None),
+        }
+    }
+
+    pub async fn get_or_refresh(
+        &self,
+        rpc_client: &Arc<RpcClient>,
+        metadata: &Arc<Metadata>,
+    ) -> Result<HashMap<String, String>> {
+        {
+            let guard = self.inner.read();
+            if let Some(cached) = guard.as_ref() {
+                if cached.cached_at.elapsed() < CACHE_TTL {
+                    return Ok(cached.to_s3_props());
+                }
+            }
+        }
+
+        self.refresh_from_server(rpc_client, metadata).await
+    }
+
+    async fn refresh_from_server(
+        &self,
+        rpc_client: &Arc<RpcClient>,
+        metadata: &Arc<Metadata>,
+    ) -> Result<HashMap<String, String>> {
+        let cluster = metadata.get_cluster();
+        let server_node = cluster
+            .get_coordinator_server()
+            .or_else(|| Some(cluster.get_one_available_server()))
+            .expect("no available server to fetch security token");
+        let conn = rpc_client.get_connection(server_node).await?;
+
+        let request = GetSecurityTokenRequest::new();
+        let response = conn.request(request).await?;
+
+        let credentials: Credentials = serde_json::from_slice(&response.token)
+            .map_err(|e| Error::JsonSerdeError(e.to_string()))?;
+
+        let mut addition_infos = HashMap::new();
+        for kv in &response.addition_info {
+            addition_infos.insert(kv.key.clone(), kv.value.clone());
+        }
+
+        let cached = CachedToken {
+            access_key_id: credentials.access_key_id,
+            secret_access_key: credentials.access_key_secret,
+            security_token: credentials.security_token,
+            addition_infos,
+            cached_at: Instant::now(),
+        };
+
+        let props = cached.to_s3_props();
+        *self.inner.write() = Some(cached);
+
+        Ok(props)
+    }
+}
+
+impl Default for CredentialsCache {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+
+
+
+
+
+