[rust] Add KvBatchScanner for full PK-table bucket scans

bindings/cpp/include/fluss.hpp

@@ -1421,6 +1421,8 @@ struct Configuration {
     size_t lookup_max_inflight_requests{128};
     // Maximum number of lookup retries
     int32_t lookup_max_retries{std::numeric_limits<int32_t>::max()};
+    // Maximum bytes to fetch per KV scanner batch request
+    int32_t scanner_kv_fetch_max_bytes{4 * 1024 * 1024};
 };
 
 class Connection {

bindings/cpp/src/lib.rs

@@ -78,6 +78,7 @@ mod ffi {
         lookup_batch_timeout_ms: u64,
         lookup_max_inflight_requests: usize,
         lookup_max_retries: i32,
+        scanner_kv_fetch_max_bytes: i32,
     }
 
     struct FfiResult {
@@ -982,6 +983,7 @@ fn new_connection(config: &ffi::FfiConfig) -> ffi::FfiPtrResult {
         lookup_batch_timeout_ms: config.lookup_batch_timeout_ms,
         lookup_max_inflight_requests: config.lookup_max_inflight_requests,
         lookup_max_retries: config.lookup_max_retries,
+        scanner_kv_fetch_max_bytes: config.scanner_kv_fetch_max_bytes,
     };
 
     let conn = RUNTIME.block_on(async { fcore::client::FlussConnection::new(config_core).await });

bindings/elixir/native/fluss_nif/src/atoms.rs

@@ -100,6 +100,10 @@ rustler::atoms! {
     ineligible_replica_exception,
     invalid_alter_table_exception,
     deletion_disabled_exception,
+    scanner_expired_exception,
+    unknown_scanner_id_exception,
+    invalid_scan_request_exception,
+    too_many_scanners,
     client_error,
 }
 
@@ -212,6 +216,10 @@ fn api_error_atom(code: i32) -> Atom {
         FlussError::IneligibleReplicaException => ineligible_replica_exception(),
         FlussError::InvalidAlterTableException => invalid_alter_table_exception(),
         FlussError::DeletionDisabledException => deletion_disabled_exception(),
+        FlussError::ScannerExpiredException => scanner_expired_exception(),
+        FlussError::UnknownScannerIdException => unknown_scanner_id_exception(),
+        FlussError::InvalidScanRequestException => invalid_scan_request_exception(),
+        FlussError::TooManyScanners => too_many_scanners(),
     }
 }
 

crates/fluss/Cargo.toml

@@ -41,6 +41,9 @@ storage-fs = ["opendal/services-fs"]
 storage-s3 = ["opendal/services-s3"]
 storage-oss = ["opendal/services-oss"]
 integration_tests = []
+# Gates tests that exercise APIs only available on Fluss 1.x servers.
+# Enable alongside `integration_tests` when running against a 1.x server image.
+fluss_v1 = []
 
 [dependencies]
 arrow = { workspace = true }