feat: support subscribe from remote

bindings/cpp/src/lib.rs

@@ -243,7 +243,7 @@ fn new_connection(bootstrap_server: &str) -> Result<*mut Connection, String> {
             }));
             Ok(conn)
         }
-        Err(e) => Err(format!("Failed to connect: {}", e)),
+        Err(e) => Err(format!("Failed to connect: {e}")),
     }
 }
 
@@ -264,7 +264,7 @@ impl Connection {
                 let admin = Box::into_raw(Box::new(Admin { inner: admin }));
                 Ok(admin)
             }
-            Err(e) => Err(format!("Failed to get admin: {}", e)),
+            Err(e) => Err(format!("Failed to get admin: {e}")),
         }
     }
 
@@ -287,7 +287,7 @@ impl Connection {
                 }));
                 Ok(table)
             }
-            Err(e) => Err(format!("Failed to get table: {}", e)),
+            Err(e) => Err(format!("Failed to get table: {e}")),
         }
     }
 }
@@ -398,7 +398,7 @@ impl Table {
 
         let table_append = match fluss_table.new_append() {
             Ok(a) => a,
-            Err(e) => return Err(format!("Failed to create append: {}", e)),
+            Err(e) => return Err(format!("Failed to create append: {e}")),
         };
 
         let writer = table_append.create_writer();
@@ -413,7 +413,10 @@ impl Table {
             self.table_info.clone(),
         );
 
-        let scanner = fluss_table.new_scan().create_log_scanner();
+        let scanner = match fluss_table.new_scan().create_log_scanner() {
+            Ok(a) => a,
+            Err(e) => return Err(format!("Failed to create log scanner: {e}")),
+        };
         let scanner = Box::into_raw(Box::new(LogScanner { inner: scanner }));
         Ok(scanner)
     }
@@ -431,9 +434,12 @@ impl Table {
         let scan = fluss_table.new_scan();
         let scan = match scan.project(&column_indices) {
             Ok(s) => s,
-            Err(e) => return Err(format!("Failed to project columns: {}", e)),
+            Err(e) => return Err(format!("Failed to project columns: {e}")),
+        };
+        let scanner = match scan.create_log_scanner() {
+            Ok(a) => a,
+            Err(e) => return Err(format!("Failed to create log scanner: {e}")),
         };
-        let scanner = scan.create_log_scanner();
         let scanner = Box::into_raw(Box::new(LogScanner { inner: scanner }));
         Ok(scanner)
     }

bindings/cpp/src/types.rs

@@ -64,7 +64,7 @@ fn ffi_data_type_to_core(dt: i32) -> Result<fcore::metadata::DataType> {
         DATA_TYPE_TIME => Ok(fcore::metadata::DataTypes::time()),
         DATA_TYPE_TIMESTAMP => Ok(fcore::metadata::DataTypes::timestamp()),
         DATA_TYPE_TIMESTAMP_LTZ => Ok(fcore::metadata::DataTypes::timestamp_ltz()),
-        _ => Err(anyhow!("Unknown data type: {}", dt)),
+        _ => Err(anyhow!("Unknown data type: {dt}")),
     }
 }
 
@@ -423,10 +423,7 @@ fn core_row_to_ffi_fields(row: &fcore::row::ColumnarRow) -> Vec<ffi::FfiDatum> {
                     datum.i32_val = array.value(row_id);
                     datum
                 }
-                _ => panic!(
-                    "Will never come here. Unsupported Time32 unit for column {}",
-                    i
-                ),
+                _ => panic!("Will never come here. Unsupported Time32 unit for column {i}"),
             },
             ArrowDataType::Time64(unit) => match unit {
                 TimeUnit::Microsecond => {
@@ -449,14 +446,10 @@ fn core_row_to_ffi_fields(row: &fcore::row::ColumnarRow) -> Vec<ffi::FfiDatum> {
                     datum.i64_val = array.value(row_id);
                     datum
                 }
-                _ => panic!(
-                    "Will never come here. Unsupported Time64 unit for column {}",
-                    i
-                ),
+                _ => panic!("Will never come here. Unsupported Time64 unit for column {i}"),
             },
             other => panic!(
-                "Will never come here. Unsupported Arrow data type for column {}: {:?}",
-                i, other
+                "Will never come here. Unsupported Arrow data type for column {i}: {other:?}"
             ),
         };
 

bindings/python/src/table.rs

@@ -67,7 +67,11 @@ impl FlussTable {
 
             let table_scan = fluss_table.new_scan();
 
-            let rust_scanner = table_scan.create_log_scanner();
+            let rust_scanner = table_scan.create_log_scanner().map_err(|e| {
+                PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(format!(
+                    "Failed to create log scanner: {e:?}"
+                ))
+            })?;
 
             let admin = conn
                 .get_admin()

crates/examples/src/example_table.rs

@@ -70,7 +70,7 @@ pub async fn main() -> Result<()> {
     try_join!(f1, f2, append_writer.flush())?;
 
     // scan rows
-    let log_scanner = table.new_scan().create_log_scanner();
+    let log_scanner = table.new_scan().create_log_scanner()?;
     log_scanner.subscribe(0, 0).await?;
 
     loop {

crates/fluss/Cargo.toml

@@ -22,6 +22,14 @@ version = { workspace = true }
 name = "fluss"
 build = "src/build.rs"
 
+[features]
+default = ["storage-memory", "storage-fs"]
+storage-all = ["storage-memory", "storage-fs"]
+
+storage-memory = ["opendal/services-memory"]
+storage-fs = ["opendal/services-fs"]
+integration_tests = []
+
 [dependencies]
 arrow = { workspace = true }
 arrow-schema = "57.0.0"
@@ -45,16 +53,17 @@ ordered-float = { version = "4", features = ["serde"] }
 parse-display = "0.10"
 ref-cast = "1.0"
 chrono = { workspace = true }
-oneshot = "0.1.11"
+opendal = "0.53.3"
+url = "2.5.7"
+async-trait = "0.1.89"
+uuid = { version = "1.10", features = ["v4"] }
+tempfile= "3.23.0"
 
 [dev-dependencies]
 testcontainers = "0.25.0"
 once_cell = "1.19"
 test-env-helpers = "0.2.2"
 
-[features]
-integration_tests = []
-
 
 [build-dependencies]
 prost-build = {  version = "0.13.5"  }

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

@@ -26,6 +26,7 @@ pub const EARLIEST_OFFSET: i64 = -2;
 
 mod append;
 
+mod remote_log;
 mod scanner;
 mod writer;