[rust] Allow configuring remote log download directory (#4603)

* [rust] Allow configuring remote log download directory

Add a configurable parent for remote log downloads, matching the Java client capability while preserving the Rust OS temporary-directory default. Each scanner owns an isolated temporary subdirectory and retains the existing cleanup lifecycle.

Expose the option in Python and C++ and document it in both API trees. Cover directory creation, isolation, cleanup, invalid paths, Unix permission errors, and binding configuration.

* [elixir] Expose the remote log download directory option

---------

Co-authored-by: liuzhuo <liuzhuo@xiaohongshu.com>

Author: spoon-lz

fluss-rust/README.md

@@ -262,6 +262,7 @@ cargo build --example example-table --release
 | `scanner_log_fetch_max_bytes`          | 16 MB             | Max bytes per fetch request                   |
 | `scanner_log_fetch_wait_max_time_ms`   | 500 ms            | Max wait time for fetch                       |
 | `scanner_remote_log_read_concurrency`  | 4                 | Concurrency for remote log reads              |
+| `scanner_io_tmp_dir`                  | OS temp dir/fluss | Parent for isolated remote log download directories |
 | `connect_timeout_ms`                   | 15 s              | Connection timeout                            |
 | `security_sasl_username` / `security_sasl_password` | — | SASL PLAIN authentication             |
 

fluss-rust/bindings/cpp/include/fluss.hpp

@@ -1645,6 +1645,8 @@ struct Configuration {
     size_t remote_file_download_thread_num{3};
     // Remote log read concurrency within one file (streaming read path)
     size_t scanner_remote_log_read_concurrency{4};
+    // Parent directory for downloaded remote logs; empty uses the OS temporary directory plus `/fluss`
+    std::string scanner_io_tmp_dir;
     // Maximum number of records returned in a single call to Poll() for LogScanner
     size_t scanner_log_max_poll_records{500};
     // Maximum bytes per fetch response for LogScanner (16 MB)

fluss-rust/bindings/cpp/src/ffi_converter.hpp

@@ -223,6 +223,7 @@ inline ffi::FfiConfig to_ffi_config(const Configuration& config) {
     ffi_config.scanner_remote_log_prefetch_num = config.scanner_remote_log_prefetch_num;
     ffi_config.remote_file_download_thread_num = config.remote_file_download_thread_num;
     ffi_config.scanner_remote_log_read_concurrency = config.scanner_remote_log_read_concurrency;
+    ffi_config.scanner_io_tmp_dir = rust::String(config.scanner_io_tmp_dir);
     ffi_config.scanner_log_max_poll_records = config.scanner_log_max_poll_records;
     ffi_config.scanner_log_fetch_max_bytes = config.scanner_log_fetch_max_bytes;
     ffi_config.scanner_log_fetch_min_bytes = config.scanner_log_fetch_min_bytes;

fluss-rust/bindings/cpp/src/lib.rs

@@ -74,6 +74,7 @@ mod ffi {
         scanner_remote_log_prefetch_num: usize,
         remote_file_download_thread_num: usize,
         scanner_remote_log_read_concurrency: usize,
+        scanner_io_tmp_dir: String,
         scanner_log_max_poll_records: usize,
         scanner_log_fetch_max_bytes: i32,
         scanner_log_fetch_min_bytes: i32,
@@ -1193,6 +1194,8 @@ fn new_connection(config: &ffi::FfiConfig) -> ffi::FfiPtrResult {
         scanner_remote_log_prefetch_num: config.scanner_remote_log_prefetch_num,
         remote_file_download_thread_num: config.remote_file_download_thread_num,
         scanner_remote_log_read_concurrency: config.scanner_remote_log_read_concurrency,
+        scanner_io_tmp_dir: (!config.scanner_io_tmp_dir.is_empty())
+            .then(|| config.scanner_io_tmp_dir.to_string()),
         scanner_log_max_poll_records: config.scanner_log_max_poll_records,
         scanner_log_fetch_max_bytes: config.scanner_log_fetch_max_bytes,
         scanner_log_fetch_min_bytes: config.scanner_log_fetch_min_bytes,

fluss-rust/bindings/cpp/test/test_ffi_converter.cpp

@@ -62,6 +62,15 @@ TEST(FfiConverterTest, KvBackpressureConfiguration) {
     EXPECT_TRUE(fluss::ErrorCode::IsRetriable(72));
 }
 
+TEST(FfiConverterTest, RemoteLogDirectoryConfiguration) {
+    fluss::Configuration config;
+    EXPECT_TRUE(config.scanner_io_tmp_dir.empty());
+    EXPECT_TRUE(fluss::utils::to_ffi_config(config).scanner_io_tmp_dir.empty());
+    config.scanner_io_tmp_dir = "/tmp/fluss logs";
+    auto ffi_config = fluss::utils::to_ffi_config(config);
+    EXPECT_EQ(std::string(ffi_config.scanner_io_tmp_dir), "/tmp/fluss logs");
+}
+
 TEST(FfiConverterTest, WriterRetryBackoffConfiguration) {
     fluss::Configuration config;
     EXPECT_EQ(config.writer_retry_backoff_ms, 100u);

fluss-rust/bindings/elixir/lib/fluss/config.ex

@@ -42,6 +42,7 @@ defmodule Fluss.Config do
             lookup_max_retries: nil,
             lookup_queue_size: nil,
             remote_file_download_thread_num: nil,
+            scanner_io_tmp_dir: nil,
             scanner_log_fetch_max_bytes: nil,
             scanner_log_fetch_max_bytes_for_bucket: nil,
             scanner_log_fetch_min_bytes: nil,
@@ -78,6 +79,7 @@ defmodule Fluss.Config do
           lookup_max_retries: non_neg_integer() | nil,
           lookup_queue_size: non_neg_integer() | nil,
           remote_file_download_thread_num: non_neg_integer() | nil,
+          scanner_io_tmp_dir: String.t() | nil,
           scanner_log_fetch_max_bytes: non_neg_integer() | nil,
           scanner_log_fetch_max_bytes_for_bucket: non_neg_integer() | nil,
           scanner_log_fetch_min_bytes: non_neg_integer() | nil,
@@ -149,6 +151,11 @@ defmodule Fluss.Config do
       when is_non_neg_integer(threads),
       do: %{config | remote_file_download_thread_num: threads}
 
+  @doc "Sets the parent directory for remote log downloads."
+  @spec set_scanner_io_tmp_dir(t(), String.t()) :: t()
+  def set_scanner_io_tmp_dir(%__MODULE__{} = config, path) when is_binary(path),
+    do: %{config | scanner_io_tmp_dir: path}
+
   @spec set_scanner_log_fetch_max_bytes(t(), non_neg_integer()) :: t()
   def set_scanner_log_fetch_max_bytes(%__MODULE__{} = config, max_bytes)
       when is_non_neg_integer(max_bytes),

fluss-rust/bindings/elixir/native/fluss_nif/src/config.rs

@@ -38,6 +38,7 @@ pub struct NifConfig {
     pub lookup_max_retries: Option<i32>,
     pub lookup_queue_size: Option<u64>,
     pub remote_file_download_thread_num: Option<u64>,
+    pub scanner_io_tmp_dir: Option<String>,
     pub scanner_log_fetch_max_bytes: Option<i32>,
     pub scanner_log_fetch_max_bytes_for_bucket: Option<i32>,
     pub scanner_log_fetch_min_bytes: Option<i32>,
@@ -70,6 +71,7 @@ impl NifConfig {
     pub fn into_core(self) -> Config {
         let mut config = Config {
             bootstrap_servers: self.bootstrap_servers,
+            scanner_io_tmp_dir: self.scanner_io_tmp_dir,
             ..Config::default()
         };
         if let Some(timeout) = self.connect_timeout_ms {
@@ -177,3 +179,64 @@ impl NifConfig {
         config
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn default_config() -> NifConfig {
+        NifConfig {
+            bootstrap_servers: "localhost:9123".to_string(),
+            connect_timeout_ms: None,
+            lookup_batch_timeout_ms: None,
+            lookup_max_batch_size: None,
+            lookup_max_inflight_requests: None,
+            lookup_max_retries: None,
+            lookup_queue_size: None,
+            remote_file_download_thread_num: None,
+            scanner_io_tmp_dir: None,
+            scanner_log_fetch_max_bytes: None,
+            scanner_log_fetch_max_bytes_for_bucket: None,
+            scanner_log_fetch_min_bytes: None,
+            scanner_log_fetch_wait_max_time_ms: None,
+            scanner_log_max_poll_records: None,
+            scanner_remote_log_prefetch_num: None,
+            scanner_remote_log_read_concurrency: None,
+            security_protocol: None,
+            security_sasl_mechanism: None,
+            security_sasl_password: None,
+            security_sasl_username: None,
+            writer_acks: None,
+            writer_batch_size: None,
+            writer_batch_timeout_ms: None,
+            writer_bucket_no_key_assigner: None,
+            writer_buffer_memory_size: None,
+            writer_buffer_wait_timeout_ms: None,
+            writer_kv_backpressure_max_throttle_ms: None,
+            writer_dynamic_batch_size_enabled: None,
+            writer_dynamic_batch_size_min: None,
+            writer_enable_idempotence: None,
+            writer_max_inflight_requests_per_bucket: None,
+            writer_request_max_size: None,
+            writer_retries: None,
+            writer_retry_backoff_ms: None,
+            writer_retry_max_backoff_ms: None,
+        }
+    }
+
+    #[test]
+    fn scanner_io_tmp_dir_is_preserved_in_core_config() {
+        for directory in [
+            None,
+            Some("/tmp/fluss logs"),
+            Some("relative/日志"),
+            Some(""),
+        ] {
+            let mut config = default_config();
+            config.scanner_io_tmp_dir = directory.map(str::to_string);
+            let core = config.into_core();
+            assert_eq!(core.scanner_io_tmp_dir.as_deref(), directory);
+            assert_eq!(core.bootstrap_servers, "localhost:9123");
+        }
+    }
+}

fluss-rust/bindings/elixir/test/config_test.exs

@@ -79,6 +79,30 @@ defmodule Fluss.ConfigTest do
     assert config.remote_file_download_thread_num == 4
   end
 
+  test "scanner_io_tmp_dir defaults to nil" do
+    assert Fluss.Config.new("localhost:9123").scanner_io_tmp_dir == nil
+    assert Fluss.Config.default().scanner_io_tmp_dir == nil
+  end
+
+  test "set_scanner_io_tmp_dir/2 preserves the configured path" do
+    for path <- ["/tmp/fluss logs", "relative/日志", ""] do
+      config =
+        Fluss.Config.new("localhost:9123")
+        |> Fluss.Config.set_scanner_io_tmp_dir(path)
+
+      assert config.scanner_io_tmp_dir == path
+      assert config.bootstrap_servers == "localhost:9123"
+    end
+  end
+
+  test "set_scanner_io_tmp_dir/2 rejects non-string values" do
+    for path <- [nil, 123, [:tmp]] do
+      assert_raise FunctionClauseError, fn ->
+        Fluss.Config.set_scanner_io_tmp_dir(Fluss.Config.default(), path)
+      end
+    end
+  end
+
   test "set_scanner_log_fetch_max_bytes/2 sets the fetch max bytes" do
     config =
       Fluss.Config.new("localhost:9123")

fluss-rust/bindings/python/fluss/__init__.pyi

@@ -202,6 +202,10 @@ class Config:
     @scanner_remote_log_read_concurrency.setter
     def scanner_remote_log_read_concurrency(self, num: int) -> None: ...
     @property
+    def scanner_io_tmp_dir(self) -> Optional[str]: ...
+    @scanner_io_tmp_dir.setter
+    def scanner_io_tmp_dir(self, path: Optional[str]) -> None: ...
+    @property
     def scanner_log_max_poll_records(self) -> int: ...
     @scanner_log_max_poll_records.setter
     def scanner_log_max_poll_records(self, num: int) -> None: ...

fluss-rust/bindings/python/src/config.rs

@@ -108,6 +108,9 @@ impl Config {
                                 ))
                             })?;
                     }
+                    "scanner.io.tmpdir" => {
+                        config.scanner_io_tmp_dir = Some(value);
+                    }
                     "scanner.log.max-poll-records" => {
                         config.scanner_log_max_poll_records =
                             value.parse::<usize>().map_err(|e| {
@@ -375,6 +378,18 @@ impl Config {
         self.inner.scanner_remote_log_read_concurrency = num;
     }
 
+    /// Parent directory for remote log downloads.
+    /// None uses the OS temporary directory plus `/fluss`.
+    #[getter]
+    fn scanner_io_tmp_dir(&self) -> Option<String> {
+        self.inner.scanner_io_tmp_dir.clone()
+    }
+
+    #[setter]
+    fn set_scanner_io_tmp_dir(&mut self, path: Option<String>) {
+        self.inner.scanner_io_tmp_dir = path;
+    }
+
     /// Get the scanner log max poll records
     #[getter]
     fn scanner_log_max_poll_records(&self) -> usize {