[rust] Support KvScan in Rust client

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

@@ -1548,6 +1548,8 @@ struct Configuration {
     int32_t scanner_log_fetch_wait_max_time_ms{500};
     // Maximum bytes per fetch response per bucket for LogScanner (1 MB)
     int32_t scanner_log_fetch_max_bytes_for_bucket{1024 * 1024};
+    // Maximum bytes per fetch response per bucket for a full KV scan (4 MB)
+    int32_t scanner_kv_fetch_max_bytes{4 * 1024 * 1024};
     int64_t writer_batch_timeout_ms{100};
     // Whether to enable idempotent writes
     bool writer_enable_idempotence{true};

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

@@ -228,6 +228,7 @@ inline ffi::FfiConfig to_ffi_config(const Configuration& config) {
     ffi_config.scanner_log_fetch_min_bytes = config.scanner_log_fetch_min_bytes;
     ffi_config.scanner_log_fetch_wait_max_time_ms = config.scanner_log_fetch_wait_max_time_ms;
     ffi_config.scanner_log_fetch_max_bytes_for_bucket = config.scanner_log_fetch_max_bytes_for_bucket;
+    ffi_config.scanner_kv_fetch_max_bytes = config.scanner_kv_fetch_max_bytes;
     ffi_config.writer_batch_timeout_ms = config.writer_batch_timeout_ms;
     ffi_config.writer_enable_idempotence = config.writer_enable_idempotence;
     ffi_config.writer_max_inflight_requests_per_bucket =

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

@@ -64,6 +64,7 @@ mod ffi {
         scanner_log_fetch_min_bytes: i32,
         scanner_log_fetch_wait_max_time_ms: i32,
         scanner_log_fetch_max_bytes_for_bucket: i32,
+        scanner_kv_fetch_max_bytes: i32,
         writer_batch_timeout_ms: i64,
         writer_enable_idempotence: bool,
         writer_max_inflight_requests_per_bucket: usize,
@@ -1143,6 +1144,7 @@ fn new_connection(config: &ffi::FfiConfig) -> ffi::FfiPtrResult {
         scanner_log_fetch_min_bytes: config.scanner_log_fetch_min_bytes,
         scanner_log_fetch_wait_max_time_ms: config.scanner_log_fetch_wait_max_time_ms,
         scanner_log_fetch_max_bytes_for_bucket: config.scanner_log_fetch_max_bytes_for_bucket,
+        scanner_kv_fetch_max_bytes: config.scanner_kv_fetch_max_bytes,
         writer_enable_idempotence: config.writer_enable_idempotence,
         writer_max_inflight_requests_per_bucket: config.writer_max_inflight_requests_per_bucket,
         writer_buffer_memory_size: config.writer_buffer_memory_size,

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

@@ -97,6 +97,10 @@ defmodule Fluss.Error do
           | :invalid_alter_table_exception
           | :deletion_disabled_exception
           | :storage_backpressure_exception
+          | :scanner_expired_exception
+          | :unknown_scanner_id_exception
+          | :invalid_scan_request_exception
+          | :too_many_scanners
           | :client_error
 
   @type t :: %__MODULE__{code: code(), error_code: integer(), message: String.t()}

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

@@ -101,6 +101,10 @@ rustler::atoms! {
     invalid_alter_table_exception,
     deletion_disabled_exception,
     storage_backpressure_exception,
+    scanner_expired_exception,
+    unknown_scanner_id_exception,
+    invalid_scan_request_exception,
+    too_many_scanners,
     client_error,
 }
 
@@ -214,6 +218,10 @@ fn api_error_atom(code: i32) -> Atom {
         FlussError::InvalidAlterTableException => invalid_alter_table_exception(),
         FlussError::DeletionDisabledException => deletion_disabled_exception(),
         FlussError::StorageBackpressureException => storage_backpressure_exception(),
+        FlussError::ScannerExpired => scanner_expired_exception(),
+        FlussError::UnknownScannerId => unknown_scanner_id_exception(),
+        FlussError::InvalidScanRequest => invalid_scan_request_exception(),
+        FlussError::TooManyScanners => too_many_scanners(),
     }
 }
 

fluss-rust/crates/fluss/src/client/metadata.rs

@@ -502,10 +502,17 @@ impl Metadata {
 #[cfg(test)]
 impl Metadata {
     pub(crate) fn new_for_test(cluster: Arc<Cluster>) -> Self {
+        Self::new_for_test_with_connections(cluster, Arc::new(RpcClient::new()))
+    }
+
+    pub(crate) fn new_for_test_with_connections(
+        cluster: Arc<Cluster>,
+        connections: Arc<RpcClient>,
+    ) -> Self {
         let (cluster_version_tx, _) = watch::channel(0);
         Metadata {
             cluster: RwLock::new(cluster),
-            connections: Arc::new(RpcClient::new()),
+            connections,
             bootstrap: Arc::from(""),
             cluster_version_tx,
         }

fluss-rust/crates/fluss/src/client/table/batch_scanner.rs

@@ -287,9 +287,10 @@ async fn decode_kv_batch(
     table_info: &TableInfo,
     schema_getter: &ClientSchemaGetter,
     projected_fields: Option<&[usize]>,
-    raw: Vec<u8>,
+    raw: impl Into<Bytes>,
     limit: usize,
 ) -> Result<RecordBatch> {
+    let raw: Bytes = raw.into();
     // No records: return an empty (projected) batch.
     if raw.is_empty() {
         return empty_record_batch(table_info.get_row_type(), projected_fields);
@@ -306,7 +307,7 @@ async fn decode_kv_batch(
             source: None,
         })?;
 
-    let batch = ValueRecordBatch::new(Bytes::from(raw));
+    let batch = ValueRecordBatch::new(raw);
     let ranges = batch.value_ranges()?;
 
     // Collect the distinct schema ids present, then build one decoder per id
@@ -455,6 +456,9 @@ fn project_batch(
     }
 }
 
+mod kv_batch_scanner;
+pub use kv_batch_scanner::{KvBatchReadOutcome, KvBatchScanner, KvSnapshotScanner};
+
 #[cfg(test)]
 mod tests {
     use super::*;